Tello EDU状态监控实战：用Python实时获取无人机飞行数据

当你第一次让Tello EDU无人机起飞，看着它平稳悬停在空中，那种感觉确实很酷。但作为一名开发者，真正让你兴奋的，可能不只是让它飞起来，而是能“看见”它飞行的每一个细节——它的速度、高度、电池还剩多少、电机是否在正常运转。这就像从一名普通的飞行员，变成了拥有全息仪表盘的机长。状态监控，正是打开这扇感知大门的钥匙。这篇文章，就是为那些不满足于基础控制，希望深入无人机数据腹地，进行飞行分析、性能评估甚至异常预警的开发者准备的。我们将绕过简单的指令发送，直击Tello EDU通过8890端口流淌出的实时数据流，用Python构建一个强大的状态监控与可视化系统。

1. 理解Tello EDU的数据通信架构

在开始写代码之前，我们必须先搞清楚Tello EDU是如何与我们对话的。很多教程只告诉你连接Wi-Fi、发送指令，却很少解释背后的通信模型，这导致一旦数据流出现问题，调试起来就无从下手。

Tello EDU采用了一种清晰的双通道UDP通信机制。你可以把它想象成两个并行的无线电频道：

• 命令/响应通道 (端口 8889)：这是一个典型的“一问一答”模式。你的程序（运行在PC或树莓派上）向Tello (192.168.10.1:8889) 发送一个字符串命令，比如 "takeoff"。Tello接收到后执行动作，然后向你的程序绑定的本地端口（例如 9000）返回一个简短的响应，通常是 "ok" 或错误信息。这个通道是你控制无人机起飞、降落、移动的基础。
• 状态流通道 (端口 8890)：这才是我们监控功能的核心。这是一个单向的、持续的数据流。一旦你连接到Tello的Wi-Fi，Tello就会自动地、以大约10Hz的频率，向你的设备广播状态数据包。你的程序不需要主动请求，只需要在本地开启一个UDP Socket，绑定到8890端口，然后“监听”这个广播即可。这个数据流包含了无人机几乎所有的实时传感器信息。

注意：状态流是广播式的。这意味着只要你设备上的某个程序绑定了8890端口，它就能收到数据。这也意味着你需要确保端口没有被其他程序占用。

为什么是UDP而不是TCP？对于实时性要求高、允许少量数据丢失的传感器数据流，UDP的无连接、低开销特性比TCP更合适。丢失一两个状态包对整体监控趋势影响不大，但TCP的握手、重传机制会带来不必要的延迟。

理解了这个架构，我们就能明白，状态监控程序的核心任务就是：建立一个稳定监听8890端口的服务，持续解析收到的数据包，并将其转化为可读、可用的信息。

2. 构建稳健的状态数据监听服务

直接修改一个简单的控制脚本，把端口从8889改成8890，确实能收到数据。但在实际项目中，我们需要一个更健壮、更模块化的监听服务。下面我们来一步步构建它。

首先，我们需要创建一个独立的Python模块来处理状态监听。我们将使用socket和threading库，让监听在后台线程中运行，不影响主程序的其他逻辑。

# tello_status_listener.py
import socket
import threading
import json
from datetime import datetime
from typing import Dict, Optional, Callable

class TelloStatusListener:
    """
    Tello EDU 状态监听器类。
    负责在后台线程中持续接收并解析来自8890端口的状态数据。
    """
    
    def __init__(self, local_ip='0.0.0.0', local_port=8890):
        """
        初始化监听器。
        :param local_ip: 本地绑定IP，通常为'0.0.0.0'以接收所有接口的数据
        :param local_port: 本地绑定端口，必须为8890以接收Tello广播
        """
        self.local_ip = local_ip
        self.local_port = local_port
        self.sock = None
        self.listening = False
        self.listener_thread = None
        self.current_status = {}  # 存储最新解析的状态字典
        self.status_callbacks = []  # 注册状态更新回调函数列表
        self._lock = threading.Lock()  # 用于线程安全地更新状态

    def start_listening(self):
        """启动状态监听线程。"""
        if self.listening:
            print("[Status Listener] 监听器已在运行。")
            return

        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        try:
            self.sock.bind((self.local_ip, self.local_port))
        except OSError as e:
            print(f"[Status Listener] 绑定端口 {self.local_port} 失败: {e}")
            print("请检查端口是否被其他程序（如另一个Python实例、其他无人机软件）占用。")
            return

        self.listening = True
        self.listener_thread = threading.Thread(target=self._listen_loop, daemon=True)
        self.listener_thread.start()
        print(f"[Status Listener] 已启动，正在监听 {self.local_ip}:{self.local_port}")

    def _listen_loop(self):
        """监听循环，运行在独立线程中。"""
        buffer_size = 1518  # 典型MTU大小
        while self.listening:
            try:
                data, _ = self.sock.recvfrom(buffer_size)
                status_text = data.decode('utf-8').strip()
                self._parse_and_update_status(status_text)
            except socket.timeout:
                continue
            except Exception as e:
                if self.listening:  # 仅在仍处于监听状态时打印错误
                    print(f"[Status Listener] 接收数据时出错: {e}")
                break

    def _parse_and_update_status(self, raw_string: str):
        """
        解析原始状态字符串并更新当前状态。
        原始字符串格式如：'pitch:12;roll:-3;yaw:45;vgx:0;vgy:0;vgz:0;...'
        """
        status_dict = {}
        try:
            for item in raw_string.split(';'):
                if ':' in item:
                    key, value = item.split(':', 1)
                    # 尝试将值转换为整数或浮点数
                    try:
                        # 处理可能的小数
                        if '.' in value:
                            status_dict[key] = float(value)
                        else:
                            status_dict[key] = int(value)
                    except ValueError:
                        # 转换失败则保留字符串
                        status_dict[key] = value
        except Exception as e:
            print(f"[Status Listener] 解析状态字符串失败: {e} | 原始数据: {raw_string}")
            return

        # 添加时间戳
        status_dict['timestamp'] = datetime.now().isoformat()

        # 线程安全地更新状态并触发回调
        with self._lock:
            self.current_status = status_dict
            for callback in self.status_callbacks:
                try:
                    callback(status_dict.copy())  # 传递副本以避免回调函数修改原始数据
                except Exception as cb_e:
                    print(f"[Status Listener] 回调函数执行出错: {cb_e}")

    def get_current_status(self) -> Dict:
        """获取当前最新的一份状态数据（线程安全）。"""
        with self._lock:
            return self.current_status.copy()

    def register_callback(self, callback: Callable[[Dict], None]):
        """
        注册一个回调函数，当状态更新时会被调用。
        :param callback: 接收一个状态字典作为参数的函数
        """
        self.status_callbacks.append(callback)

    def stop_listening(self):
        """停止监听线程并清理资源。"""
        self.listening = False
        if self.sock:
            self.sock.close()
        if self.listener_thread:
            self.listener_thread.join(timeout=2.0)
        print("[Status Listener] 已停止。")

这个TelloStatusListener类提供了一个清晰的接口。在你的主程序中，你可以这样使用它：

# main.py
from tello_status_listener import TelloStatusListener
import time

def my_status_handler(status):
    """自定义状态处理函数示例"""
    if 'bat' in status:
        print(f"[回调] 电池电量: {status['bat']}%", end=' | ')
    if 'h' in status:
        print(f"相对高度: {status['h']} cm")

# 创建并启动监听器
listener = TelloStatusListener()
listener.start_listening()

# 注册回调函数，每当新数据到达就会自动调用
listener.register_callback(my_status_handler)

# 主程序可以做其他事情，比如发送控制指令
try:
    while True:
        # 也可以主动获取当前状态
        current = listener.get_current_status()
        if current:
            # 这里可以添加其他逻辑，比如判断电量低时报警
            pass
        time.sleep(0.1)  # 避免CPU空转
except KeyboardInterrupt:
    print("\n用户中断。")
finally:
    listener.stop_listening()  # 确保程序退出前清理资源

这种设计将状态监听与业务逻辑解耦，使得代码更易于维护和扩展。回调机制允许你在数据到达时立即触发特定动作，比如记录日志或更新UI。

3. 深度解析状态数据：从原始字符串到可操作信息

Tello EDU发送的状态字符串看起来像一长串分号分隔的键值对。直接看原始数据可能有些晦涩，我们需要一个“解码器”。官方SDK文档列出了主要参数，但结合实践，我们可以更深入地理解它们。

下面这个表格整理了最核心的状态参数及其在实际应用中的意义：

参数键	数据类型	单位/范围	含义解析与实战意义
pitch	整数	度 (-90~90)	俯仰角。机头向下为正。实战：值持续为正且较大，可能表示无人机正在逆风飞行或电机推力不平衡。
roll	整数	度 (-90~90)	横滚角。向右倾斜为正。实战：飞行中非零的恒定值，可能表示存在侧风或需要校准IMU。
yaw	整数	度 (-180~180)	偏航角。顺时针旋转为正。实战：结合角速度agz，可以判断无人机是否在稳定悬停（yaw变化慢，agz接近0）。
vgx, vgy, vgz	整数	cm/s	地速。在X（前后）、Y（左右）、Z（上下）方向的速度。注意：这是基于视觉流传感器计算的速度，在光照不足或纹理单一的地面可能不准。实战：vgz持续为负但高度h不变，可能表示无人机正在下降气流中努力维持高度。
agx, agy, agz	整数	度/秒	角速度。绕X、Y、Z轴旋转的速率。比姿态角pitch/roll/yaw更能反映瞬时的机动剧烈程度。实战：agx或agy突然大幅波动，可能提示遇到了突风或发生碰撞。
h	整数	厘米	相对起飞点的高度。由气压计和TOF（飞行时间）传感器融合得出。实战：在室内，TOF数据更可靠；室外无风环境下，气压计数据更平滑。两者差异过大可能提示传感器故障。
bat	整数	百分比	电池剩余电量。这是最重要的监控参数之一。实战：电量下降曲线并非线性。通常从100%到30%比较平缓，30%以下会加速下降。建议设置预警（如低于25%）和紧急返航（如低于15%）逻辑。
temph	整数	摄氏度	最高电机温度。实战：长时间大推力飞行（如快速爬升、载重）会导致温度升高。持续高于70°C需要警惕，可能触发过热保护导致动力下降。
tof	整数	厘米	TOF传感器测量的高度。测量范围有限（约30-1000cm），但非常精确。实战：与h对比，可以评估气压计的漂移情况。在低空避障应用中，tof是主要数据源。
time	整数	秒	电机累计运行时间。从本次开机开始计算。实战：用于估算单次飞行续航，结合电池电量，可以建立电池健康度模型。

解析完这些参数，我们可以在监听器中增强解析函数，直接计算出一些衍生指标，比如：

• 飞行平稳度：计算短时间内pitch, roll, agx, agy的方差。
• 电量消耗速率：根据bat和time的变化计算每分钟耗电百分比。
• 高度保持精度：计算h值围绕目标高度的标准差。

将这些解析逻辑集成到_parse_and_update_status方法中，你就能获得更丰富、更直接可用的监控指标。

4. 实现实时数据可视化与异常监测

有了稳定流淌的数据和清晰的含义，下一步就是让数据“说话”。可视化不仅能直观展示飞行状态，更是发现异常模式的有力工具。我们将使用Python中流行的matplotlib库，创建一个动态更新的仪表盘。

这里我们设计一个包含四个子图的实时监控仪表板：

1. 姿态仪表：用类似飞机姿态仪的方式显示pitch和roll。
2. 高度与速度趋势图：显示h（高度）和vgz（垂直速度）随时间的变化。
3. 电池与温度仪表：用进度条和数字显示bat和temph。
4. 数据表格：实时刷新所有关键参数的数值。

# visualizer.py
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import matplotlib.patches as patches
import numpy as np
from collections import deque
from tello_status_listener import TelloStatusListener  # 导入我们之前写的监听器

class TelloStatusVisualizer:
    def __init__(self, status_listener, history_len=200):
        self.listener = status_listener
        self.history_len = history_len  # 历史数据记录长度
        
        # 初始化数据历史队列
        self.time_history = deque(maxlen=history_len)
        self.h_history = deque(maxlen=history_len)  # 高度历史
        self.vgz_history = deque(maxlen=history_len) # 垂直速度历史
        self.bat_history = deque(maxlen=history_len) # 电量历史
        
        # 创建图形和子图
        self.fig, ((self.ax_att, self.ax_alt), 
                   (self.ax_bat, self.ax_table)) = plt.subplots(2, 2, figsize=(14, 10))
        self.fig.suptitle('Tello EDU 实时状态监控仪表板', fontsize=16)
        
        self._setup_attitude_display()   # 设置姿态显示
        self._setup_altitude_chart()     # 设置高度图表
        self._setup_battery_gauge()      # 设置电池仪表
        self._setup_status_table()       # 设置数据表格
        
        # 注册回调，当有新数据时更新图表
        self.listener.register_callback(self.on_new_status)
        
        # 动画对象
        self.ani = None
        
    def _setup_attitude_display(self):
        """设置飞机姿态指示器（人工地平线）。"""
        self.ax_att.set_title('飞行姿态 (Pitch & Roll)')
        self.ax_att.set_xlim(-50, 50)
        self.ax_att.set_ylim(-50, 50)
        self.ax_att.set_aspect('equal')
        self.ax_att.grid(True, linestyle='--', alpha=0.7)
        self.ax_att.axhline(y=0, color='black', linewidth=1)
        self.ax_hist.axvline(x=0, color='black', linewidth=1)
        
        # 绘制一个简单的飞机符号
        self.aircraft_body, = self.ax_att.plot([-10, 10, 0, -10], [0, 0, 8, 0], 'k-', linewidth=3)
        self.attitude_text = self.ax_att.text(-45, 45, '', fontsize=10, 
                                               bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8))
        
    def _setup_altitude_chart(self):
        """设置高度和垂直速度时间序列图。"""
        self.ax_alt.set_title('高度与垂直速度趋势')
        self.ax_alt.set_xlabel('时间 (相对)')
        self.ax_alt.set_ylabel('高度 (cm)', color='tab:blue')
        self.line_h, = self.ax_alt.plot([], [], 'b-', label='高度 (h)', linewidth=2)
        self.ax_alt.tick_params(axis='y', labelcolor='tab:blue')
        self.ax_alt.grid(True, alpha=0.3)
        
        # 创建第二个Y轴用于速度
        self.ax_vgz = self.ax_alt.twinx()
        self.ax_vgz.set_ylabel('垂直速度 (cm/s)', color='tab:red')
        self.line_vgz, = self.ax_vgz.plot([], [], 'r-', label='垂直速度 (vgz)', linewidth=1.5, alpha=0.8)
        self.ax_vgz.tick_params(axis='y', labelcolor='tab:red')
        
        # 合并图例
        lines = [self.line_h, self.line_vgz]
        labels = [l.get_label() for l in lines]
        self.ax_alt.legend(lines, labels, loc='upper left')
        
    def _setup_battery_gauge(self):
        """设置电池电量和温度仪表。"""
        self.ax_bat.set_title('电池与温度状态')
        self.ax_bat.set_xlim(0, 100)
        self.ax_bat.set_ylim(0, 3)
        self.ax_bat.axis('off')
        
        # 电池外框
        battery_rect = patches.Rectangle((10, 1), 80, 1.2, linewidth=2, 
                                         edgecolor='black', facecolor='none')
        battery_cap = patches.Rectangle((90, 1.4), 5, 0.4, linewidth=2,
                                        edgecolor='black', facecolor='black')
        self.ax_bat.add_patch(battery_rect)
        self.ax_bat.add_patch(battery_cap)
        
        # 电量填充（动态更新）
        self.battery_fill = patches.Rectangle((11, 1.1), 0, 1, 
                                              edgecolor='none', facecolor='lime')
        self.ax_bat.add_patch(self.battery_fill)
        
        # 电池文本
        self.bat_text = self.ax_bat.text(50, 2.5, '电池: --%', ha='center', fontsize=12, fontweight='bold')
        self.temp_text = self.ax_bat.text(50, 0.5, '温度: --°C', ha='center', fontsize=12)
        
    def _setup_status_table(self):
        """设置关键参数数据表格。"""
        self.ax_table.axis('tight')
        self.ax_table.axis('off')
        self.table_data = [
            ['姿态角', 'Pitch:', '--°', 'Roll:', '--°', 'Yaw:', '--°'],
            ['速度', 'VGX:', '--', 'VGY:', '--', 'VGZ:', '--'],
            ['角速度', 'AGX:', '--', 'AGY:', '--', 'AGZ:', '--'],
            ['传感器', 'TOF:', '-- cm', 'Time:', '-- s', 'Temp:', '--°C']
        ]
        self.table = self.ax_table.table(cellText=self.table_data, 
                                         cellLoc='left', 
                                         loc='center',
                                         colWidths=[0.15, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1])
        self.table.auto_set_font_size(False)
        self.table.set_fontsize(10)
        self.table.scale(1.2, 2)
        
    def on_new_status(self, status):
        """监听器回调函数：接收新状态并更新历史数据。"""
        from datetime import datetime
        current_time = datetime.now()
        
        # 记录时间戳
        self.time_history.append(current_time)
        
        # 更新各项历史数据
        for key, history, default in [
            ('h', self.h_history, 0),
            ('vgz', self.vgz_history, 0),
            ('bat', self.bat_history, 0)
        ]:
            history.append(status.get(key, default))
            
        # 存储最新状态用于表格更新
        self.latest_status = status
        
    def update_plots(self, frame):
        """动画更新函数，由FuncAnimation调用。"""
        if not hasattr(self, 'latest_status'):
            return
            
        status = self.latest_status
        
        # 1. 更新姿态显示
        pitch = status.get('pitch', 0)
        roll = status.get('roll', 0)
        # 简化：根据roll和pitch旋转飞机符号（这里用偏移模拟）
        self.aircraft_body.set_data([-10+roll/5, 10+roll/5, 0+roll/5, -10+roll/5], 
                                     [0+pitch/5, 0+pitch/5, 8+pitch/5, 0+pitch/5])
        self.attitude_text.set_text(f'Pitch: {pitch}°\nRoll: {roll}°')
        
        # 2. 更新高度/速度图
        if len(self.time_history) > 1:
            # 将时间转换为相对秒数
            time_ref = list(self.time_history)
            time_secs = [(t - time_ref[0]).total_seconds() for t in time_ref]
            
            self.line_h.set_data(time_secs, list(self.h_history))
            self.line_vgz.set_data(time_secs, list(self.vgz_history))
            self.ax_alt.relim()
            self.ax_alt.autoscale_view()
            self.ax_vgz.relim()
            self.ax_vgz.autoscale_view()
            
        # 3. 更新电池仪表
        bat = status.get('bat', 0)
        bat_width = max(0, min(78, 78 * bat / 100))  # 计算填充宽度
        self.battery_fill.set_width(bat_width)
        # 根据电量改变颜色
        if bat > 50:
            self.battery_fill.set_facecolor('lime')
        elif bat > 20:
            self.battery_fill.set_facecolor('orange')
        else:
            self.battery_fill.set_facecolor('red')
            
        self.bat_text.set_text(f'电池: {bat}%')
        
        temp = status.get('temph', 0)
        self.temp_text.set_text(f'温度: {temp}°C')
        
        # 4. 更新数据表格
        # 只更新数值单元格
        update_map = {
            'Pitch:': f'{status.get("pitch", 0)}°',
            'Roll:': f'{status.get("roll", 0)}°',
            'Yaw:': f'{status.get("yaw", 0)}°',
            'VGX:': f'{status.get("vgx", 0)}',
            'VGY:': f'{status.get("vgy", 0)}',
            'VGZ:': f'{status.get("vgz", 0)}',
            'AGX:': f'{status.get("agx", 0)}',
            'AGY:': f'{status.get("agy", 0)}',
            'AGZ:': f'{status.get("agz", 0)}',
            'TOF:': f'{status.get("tof", 0)} cm',
            'Time:': f'{status.get("time", 0)} s',
            'Temp:': f'{status.get("temph", 0)}°C'
        }
        
        for row in self.table_data:
            for i, cell in enumerate(row):
                if cell in update_map:
                    row[i+1] = update_map[cell]  # 更新下一个单元格（数值）
        
        # 刷新表格
        self.table = self.ax_table.table(cellText=self.table_data, 
                                         cellLoc='left', 
                                         loc='center',
                                         colWidths=[0.15, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1])
        self.table.auto_set_font_size(False)
        self.table.set_fontsize(10)
        self.table.scale(1.2, 2)
        
        self.fig.canvas.draw_idle()
        return self.aircraft_body, self.line_h, self.line_vgz, self.battery_fill, self.bat_text, self.temp_text
        
    def start_visualization(self):
        """启动可视化动画。"""
        self.ani = FuncAnimation(self.fig, self.update_plots, interval=200, cache_frame_data=False) # 200ms更新一次
        plt.tight_layout()
        plt.show()
        
    def stop_visualization(self):
        """停止可视化。"""
        if self.ani:
            self.ani.event_source.stop()
        plt.close(self.fig)

# 主程序示例
if __name__ == "__main__":
    listener = TelloStatusListener()
    listener.start_listening()
    
    visualizer = TelloStatusVisualizer(listener)
    
    try:
        print("启动状态监控可视化仪表板...")
        print("请确保Tello EDU已开机并连接到其Wi-Fi。")
        visualizer.start_visualization()
    except KeyboardInterrupt:
        print("\n正在关闭...")
    finally:
        visualizer.stop_visualization()
        listener.stop_listening()

运行这个可视化脚本，你会看到一个动态更新的仪表盘。它能让你在无人机飞行时，直观地看到姿态变化、高度曲线、电量消耗，所有关键参数一目了然。更重要的是，你可以基于这个框架，轻松添加异常监测逻辑。例如，在on_new_status回调函数中加入：

def check_anomalies(self, status):
    """简单的异常检测示例"""
    alerts = []
    
    # 低电量警报
    if status.get('bat', 100) < 20:
        alerts.append(f"⚠️ 低电量警报: {status['bat']}%")
        
    # 高温警报
    if status.get('temph', 0) > 70:
        alerts.append(f"🔥 电机高温警报: {status['temph']}°C")
        
    # 高度异常波动（简单示例：短时间内变化过大）
    if len(self.h_history) > 10:
        recent_heights = list(self.h_history)[-10:]
        if max(recent_heights) - min(recent_heights) > 50: # 最近10个点波动超过50cm
            alerts.append("📈 高度异常波动")
            
    # 姿态角过大（可能表示强风或失控）
    if abs(status.get('pitch', 0)) > 30 or abs(status.get('roll', 0)) > 30:
        alerts.append("✈️ 姿态角过大，请检查飞行环境")
        
    if alerts:
        # 这里可以将警报打印到控制台、记录到文件或发送到UI
        print(f"[警报] {datetime.now().strftime('%H:%M:%S')} - {' | '.join(alerts)}")

将check_anomalies注册为另一个回调函数，你的监控系统就具备了基本的异常检测能力。在实际部署中，你可以将这些警报与自动控制逻辑结合，比如触发自动返航或降落。

5. 数据持久化与离线分析

实时监控很重要，但事后分析同样有价值。将飞行数据保存下来，可以用于性能评估、故障复盘或机器学习模型训练。我们可以扩展监听器，增加数据记录功能。

一个简单有效的方法是使用CSV格式记录，因为它易于生成，也方便用Excel、Python pandas或任何数据分析工具打开。

# 在TelloStatusListener类中添加数据记录功能
import csv
from pathlib import Path

class TelloStatusListenerWithLogging(TelloStatusListener):
    """带数据记录功能的状态监听器"""
    
    def __init__(self, log_dir="flight_logs", *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(exist_ok=True)
        self.log_file = None
        self.csv_writer = None
        
    def start_listening(self):
        # 在开始监听前创建日志文件
        from datetime import datetime
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.log_file = self.log_dir / f"tello_flight_{timestamp}.csv"
        
        # 先写入CSV表头
        with open(self.log_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            # 定义要记录的字段（可以根据需要调整）
            header = ['timestamp', 'pitch', 'roll', 'yaw', 
                     'vgx', 'vgy', 'vgz', 'agx', 'agy', 'agz',
                     'h', 'bat', 'temph', 'tof', 'time']
            writer.writerow(header)
            
        # 打开文件用于追加
        self.log_file_handle = open(self.log_file, 'a', newline='', encoding='utf-8')
        self.csv_writer = csv.writer(self.log_file_handle)
        
        # 注册一个回调，将每条状态记录到CSV
        self.register_callback(self._log_status)
        
        # 调用父类的启动方法
        super().start_listening()
        print(f"[Status Listener] 数据将记录到: {self.log_file}")
        
    def _log_status(self, status):
        """将状态数据写入CSV文件"""
        if self.csv_writer:
            try:
                row = [
                    status.get('timestamp', ''),
                    status.get('pitch', 0),
                    status.get('roll', 0),
                    status.get('yaw', 0),
                    status.get('vgx', 0),
                    status.get('vgy', 0),
                    status.get('vgz', 0),
                    status.get('agx', 0),
                    status.get('agy', 0),
                    status.get('agz', 0),
                    status.get('h', 0),
                    status.get('bat', 0),
                    status.get('temph', 0),
                    status.get('tof', 0),
                    status.get('time', 0)
                ]
                self.csv_writer.writerow(row)
                self.log_file_handle.flush()  # 确保数据及时写入磁盘
            except Exception as e:
                print(f"[Status Listener] 记录数据失败: {e}")
                
    def stop_listening(self):
        """停止监听并关闭日志文件"""
        if hasattr(self, 'log_file_handle') and self.log_file_handle:
            self.log_file_handle.close()
        super().stop_listening()

有了CSV日志文件，飞行结束后，你可以用pandas进行深入分析：

# flight_analysis.py
import pandas as pd
import matplotlib.pyplot as plt

# 加载飞行日志
df = pd.read_csv('flight_logs/tello_flight_20231027_143022.csv', parse_dates=['timestamp'])

# 计算飞行时间
flight_duration = (df['timestamp'].iloc[-1] - df['timestamp'].iloc[0]).total_seconds()
print(f"飞行时长: {flight_duration:.1f} 秒")

# 计算平均功耗（假设起始电量为100%）
if not df.empty:
    start_bat = df['bat'].iloc[0]
    end_bat = df['bat'].iloc[-1]
    bat_used = start_bat - end_bat
    power_rate = bat_used / flight_duration if flight_duration > 0 else 0
    print(f"电量消耗: {bat_used:.1f}% ({power_rate:.3f}%/秒)")

# 分析高度稳定性
mean_altitude = df['h'].mean()
altitude_std = df['h'].std()
print(f"平均高度: {mean_altitude:.1f} cm, 标准差: {altitude_std:.1f} cm (越小越稳定)")

# 绘制飞行剖面图
fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True)

# 高度剖面
axes[0].plot(df['timestamp'], df['h'], 'b-', linewidth=1)
axes[0].set_ylabel('高度 (cm)')
axes[0].grid(True, alpha=0.3)
axes[0].set_title('飞行高度剖面')

# 电池曲线
axes[1].plot(df['timestamp'], df['bat'], 'g-', linewidth=2)
axes[1].set_ylabel('电池电量 (%)')
axes[1].grid(True, alpha=0.3)
axes[1].axhline(y=20, color='r', linestyle='--', alpha=0.5, label='低电量线 (20%)')
axes[1].legend()

# 姿态角变化
axes[2].plot(df['timestamp'], df['pitch'], 'r-', label='Pitch', alpha=0.7)
axes[2].plot(df['timestamp'], df['roll'], 'b-', label='Roll', alpha=0.7)
axes[2].plot(df['timestamp'], df['yaw'], 'g-', label='Yaw', alpha=0.7)
axes[2].set_ylabel('角度 (度)')
axes[2].set_xlabel('时间')
axes[2].grid(True, alpha=0.3)
axes[2].legend()

plt.tight_layout()
plt.show()

# 生成飞行报告摘要
print("\n=== 飞行报告摘要 ===")
print(f"最大高度: {df['h'].max():.0f} cm")
print(f"最小高度: {df['h'].min():.0f} cm")
print(f"最大俯仰角: {df['pitch'].abs().max():.1f}°")
print(f"最大横滚角: {df['roll'].abs().max():.1f}°")
print(f"最高电机温度: {df['temph'].max():.0f}°C")

这种从实时监控到数据记录，再到离线分析的完整流程，构成了一个专业无人机开发项目的数据闭环。它让你不仅能实时感知无人机状态，还能积累数据、量化性能、优化飞行策略。在我自己的几个Tello EDU集群项目中，正是依靠这套监控系统，才准确发现了其中一台无人机电机老化导致的功耗异常，避免了它在飞行任务中突然掉电。