简介:这是一份面向Python开发者的ADS-B消息处理工具包,基于adsb库提供对广播式自动相关监视(ADS-B)信号的解析与处理功能,适用于航空数据采集、无人机监管、机场场面监控等场景,也适合希望学习1090MHz扩展电文协议和Python消息解析的入门及中级开发者。压缩包内共49个文件,包含23个Python脚本(核心模块、客户端服务端实现及测试用例)、8个RST文档(API参考、变更日志与贡献指南)、5个TXT飞行数据日志样例,并配有YML持续集成配置、Makefile构建脚本和License许可文件,整体仅55KB,轻量便携、目录规划清晰。目前已有1016人参与学习,项目虽处开发初期,但已提供模块级API文档,便于快速上手。读者通过源码和测试样例可了解adsb包如何解析原始ADS-B消息、组织JSON输出以及搭建客户端/服务端;文档与配置模板还能帮助二次开发时快速建立测试、打包和文档生成环境,是一份实用的Python航空数据工具参考。
1. 用 Python 抓飞机:ADS-B 工具到底能帮你做什么
如果你第一次听说 ADS-B,可以先把它理解成“飞机主动广播自己在哪、飞多快、要去哪”的一套公开协议。任何一架支持 ADS-B 的飞机,都会周期性向外发送包含经纬度、高度、速度、呼号等信息的报文,地面用一台接收机就能截获并解码。而今天聊的这个 Python 工具,就是把这套接收、解码、分析流程打包成可直接调用的库,让你不用从零去啃 1090MHz 信号的比特级协议,十几行代码就能把附近的航班动态拉进自己的脚本里。我最初接触它是因为想在一个自动化看板里实时显示头顶过境的飞机,本以为要折腾射频前端和信号处理,结果发现核心路径只缺一个“能看懂 ADS-B 数据”的 Python 层。这个工具适合三类人:一是玩 RTL-SDR 但没有精力自己写解码器的业余无线电爱好者,二是做航空数据可视化或轨迹分析的应用开发者,三是想低成本验证 ADS-B 数据形态的数据工程师。
2. 抓包之前先懂协议:1090MHz 报文与 Python 的接入姿势
2.1 报文长什么样:Mode S 与位置消息的解码路径
ADS-B 的数据广播主要工作在 1090MHz 频段,采用 Mode S 扩展电文(Extended Squitter)格式。每一条报文长度固定为 112 比特,去掉前导码后,剩下一部分是 DF(Downlink Format,下行格式)字段和 ICAO 地址(飞机的全球唯一编号),再往后才是真正承载数据的 ME(Message Extended)字段。DF 值不同,报文类型就不同,常见的有 DF17(ADS-B 位置/速度消息)、DF18(TIS-B 地面广播)和 DF4/DF5/DF20/DF21(应答机消息)。位置报文里用 CPR(Compact Position Reporting,紧凑位置报告)算法把经纬度拆成奇偶两帧交替发送,单帧解不出来完整坐标,必须把连续两帧配对才能还原。这个设计是为了省带宽——把两个 35 比特以内的数值组合起来表达完整的经纬度信息。
如果从零开始解析这些比特,你需要手动做 CRC 校验、ICAO 地址查表、CPR 奇偶帧配对、高度编码转换,工程量不小。而这个 Python 工具的核心价值就在于它把这一整套解码做成了黑匣子:你给它原始数据,它返回结构化对象。常见的输入格式有两种,一种是 RTL-SDR 直接采样的 IQ 信号(需要先做幅度解调),另一种是 dump1090 这类接收软件已经解码好的 JSON/AIS 数据。实操中绝大多数人会选后者,因为省掉大量信号处理环节。下面代码演示的是最典型的接入方式——用 Python 建立 TCP 连接,从 dump1090 的对外端口读取已经解码好的 JSON 流。
import socket import json # 连接到本机 dump1090 的 30003 端口 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(("127.0.0.1", 30003)) buffer = "" while True: data = sock.recv(4096) if not data: break buffer += data.decode("utf-8", errors="ignore") # 30003 端口输出的是一行一条 JSON,按换行符切分 lines = buffer.split("\n") buffer = lines[-1] for line in lines[:-1]: if line.strip(): msg = json.loads(line) print(msg)这段代码的逻辑很直接:建立 TCP 连接后,用 recv 持续读取数据流,按换行符把一条条 JSON 解析出来。参数方面,30003 是 dump1090 默认的 raw 输出端口之一,除了它还有 30002(原始二进制)和 30005(beast 格式)。要注意的是 recv 的缓冲区大小不是你消息的上限——TCP 是流式协议,一条消息可能被拆成多个小块到达,也可能多条一起到达,所以必须用 buffer 累积数据再做切分,否则你会看到频繁的 JSON 解析错误。初次跑通时建议把 msg 原样打印出来,先看看字段长什么样,再决定后续要提取哪些信息。
2.2 数据源选型:直接收 SDR 信号 vs 对接 dump1090
决定用哪种数据源之前,得先弄清楚你的落地场景。如果你是想放在树莓派或一台长期运行的 Linux 主机上,RTL-SDR 加天线直接收 1090MHz 信号是最完整的方案,数据延迟最低,也完全自主可控。但这条路要求你先把 RTL-SDR 的驱动装好,再通过 rtl-sdr 提供的软解调器把射频信号变成可解格式,中间涉及采样率设置、增益调节、滤波,每一步都可能成为新手劝退点。相比之下,对接 dump1090 的思路是让 dum1090 负责射频接收和 Mode S 解码,Python 只处理已经结构化的 JSON 数据,工程复杂度大幅下降。
还有一种方式是连接公共服务平台获取远程数据,比如有些网站或平台会开放 API 或数据流。但请注意,这里有个坑:如果你在配置里用了“代理”之类的转发设置,先确认本地网络是否允许直连这些服务,有些公共数据流在国内网络环境下访问不稳定,你以为是工具问题,其实是链路问题。我一般不建议把远程数据流作为唯一数据源——调试周期太长,且对方随时可能改协议。最稳的组合是本地 RTL-SDR 接收做主力,公共数据流做补充校验。下表对比了三种数据源的适用场景:
| 数据源 | 延迟 | 部署难度 | 覆盖范围 | 适用场景 |
|---|---|---|---|---|
| RTL-SDR 直接接收 | 秒级 | 高 | 本地天线视距内 | 强实时、离线部署 |
| dump1090 中间层 | 秒级 | 中 | 本地天线视距内 | 快速原型、轻量应用 |
| 公共服务 API | 秒到分钟级 | 低 | 大范围 | 区域态势、备份校验 |
2.3 连接 dump1090:TCP 套接字读取原始 JSON 流
第 2.1 小节的代码是能跑通的最简版本,但真正写生产脚本时还会遇到几个细节问题:长时间运行时 socket 可能断开、数据字段类型不统一、某些航班的字段缺失导致 json.loads 报错。下面这个改良版加了断线重连和字段容错:
import socket import json import time def connect_dump1090(host="127.0.0.1", port=30003, retry_interval=5): while True: try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((host, port)) sock.settimeout(60) return sock except (socket.timeout, ConnectionRefusedError, OSError) as e: print(f"连接失败: {e}, {retry_interval}秒后重试...") time.sleep(retry_interval) sock = connect_dump1090() buffer = "" while True: try: data = sock.recv(4096) if not data: raise ConnectionResetError("连接已被关闭") buffer += data.decode("utf-8", errors="ignore") lines = buffer.split("\n") buffer = lines[-1] for line in lines[:-1]: if not line.strip(): continue try: msg = json.loads(line) # 使用 get 而不是直接索引,避免字段缺失导致崩溃 flight = msg.get("flight", "").strip() lat = msg.get("lat") lon = msg.get("lon") if lat is not None and lon is not None: print(f"{flight or '未知呼号'}: ({lat:.4f}, {lon:.4f})") except json.JSONDecodeError: # 只打印消息头部做排查,不中断主循环 print(f"解析失败,原始片段前80字符: {line[:80]}") except (socket.timeout, ConnectionResetError, OSError) as e: print(f"连接异常: {e}, 重新连接...") sock.close() sock = connect_dump1090() buffer = ""这里的核心改动有四个。一是把 socket 连接封装成函数,断线后自动重连,适配长时间挂机的场景;二是给 socket 加了 60 秒超时,配合 try-except 捕获超时异常,否则 recv 可能一直阻塞到天荒地老;三是用 get() 访问字典字段,因为并非每条报文都包含呼号、经纬度——位置消息本来就是分奇偶帧交替到达的,单帧里只有其中一种坐标信息;四是 JSON 解析失败时只打印片段不中断主循环,这在数据流混入非 JSON 噪声时至关重要。参数里 retry_interval 控制重连频率,设太短会在 dump1090 重启时疯狂刷日志,设太长又会导致断流期间数据缺失,我一般是 5 秒,这个值在工程上比较稳妥。
3. 安装与接入 dump1090:环境准备、台站配置与数据流打通
3.1 环境准备:Python 版本、RTL-SDR 驱动与 dump1090 安装
先明确一个分工:RTL-SDR 负责射频采样,dump1090 负责解调和解码,Python 工具负责读取解码结果。三者的环境配置是递进关系。Python 版本建议 3.8 以上,因为后续会用到 dataclass 和 typing 等特性,老版本不是不能用,但没必要和年份较劲。RTL-SDR 驱动的核心是让操作系统识别 RTL2832U 芯片的 USB 设备并暴露为软件定义无线电接口,Linux 下一般通过安装 librtlsdr 和 rtl-sdr 工具链完成,Windows 下常见的是 Zadig 驱动替换方案。dump1090 的安装有三种主流方式:源码编译、发行版仓库安装、Docker 镜像运行。
# Ubuntu/Debian 系安装依赖和 dump1090 sudo apt-get update sudo apt-get install -y librtlsdr-dev rtl-sdr build-essential git git clone https://github.com/flightaware/dump1090.git cd dump1090 make # 使用 Docker 的方式更省事,可跳过编译环节 docker pull ghcr.io/flightaware/dump1090:latest源码编译和 Docker 各有适用场景。源码编译能拿到最新版本,同时可以加调试参数,但对依赖库版本敏感,gcc 版本不对可能在链接阶段报错;Docker 方式隔离了环境依赖,只要宿主机能访问 USB 设备并映射端口即可,代价是调试内部状态要多绕一层。我个人的经验是:开发机先用源码编译,因为方便改代码打断点;部署到生产环境时改用 Docker,升级和回滚都干净。参数方面需要注意 librtlsdr 和 rtl-sdr 的版本匹配,新版 rtl-sdr 对驱动 API 做了调整,不匹配时会出现启动 dump1090 后报“unable to open device”的错误。
3.2 启动 dump1090:关键参数与网口输出
dump1090 的启动参数决定了它向外暴露什么数据、监听哪个端口、输出采样率多少。这条命令是我常用的组合:
dump1090 --device 0 --gain 49.6 --ifile /dev/rtl-sdr --net --net-only \ --net-http-port 8080 --net-ri-port 30003 --fix --mlat \ --ppm 0 --oversample逐个参数拆开说一下。“--device 0”指定使用第一个 RTL-SDR 设备,如果你插了多个接收器,通过 rtl_test 可以确认设备编号。“--gain 49.6”是固定增益模式,因为自动增益控制在天线周围存在强干扰源时会把增益拉到极低,导致弱信号被淹没;49.6 这个值是我在开阔环境下反复试出来的,城市密集楼宇环境建议从 40 起步逐渐往上加。“--net --net-only”组合的含义是“启动网络服务但仅通过网络输出”,这样 dump1090 不占用终端持续打印接收日志,适合后台运行。“--net-http-port 8080”提供 Web 界面端口,能看到当前接收的飞机数量、信号质量等信息,调试时非常有用。“--net-ri-port 30003”就是前面 Python 脚本连接的端口。“--fix”开启 CRC 错误修正,能提高弱信号下的解码成功率。“--mlat”启用多点定位支持,当多台接收站同步时可以通过 TDOA 定位无 ADS-B 输出的飞机。“--ppm 0”是晶振频率偏移补偿,单位是 ppm(parts per million),如果基站速度显示偏高或偏低,用 rtl_test 测出实际偏移再填入更正。
# 验证 dump1090 是否正常输出数据 curl http://127.0.0.1:8080/data/aircraft.json这条 curl 命令返回的 JSON 里有 now(当前 Unix 时间戳)、messages(累计解码消息数)、aircraft(当前所有被跟踪飞机的数组)三个核心字段。如果 aircraft 数组长度大于 0,说明你已经成功解出了飞机报文。注意 messages 这个字段会持续增长,如果一段时间内 messages 增加但 aircraft 始终为空,那大概率是 CRC 校验失败率过高,或者天线增益设置不当导致误码率上升,而不是没有飞机经过。
3.3 在 Python 里验证数据流:能打印出飞机才算入门
前面已经搭好了链路,现在用一段简洁脚本验证端到端是否真正跑通。这一步别着急写复杂逻辑,先让屏幕上有飞机出现再说。
import urllib.request import json # 从 dump1090 的 HTTP 接口读取实时飞机数据 url = "http://127.0.0.1:8080/data/aircraft.json" with urllib.request.urlopen(url, timeout=10) as resp: data = json.loads(resp.read().decode("utf-8")) aircraft_list = data.get("aircraft", []) print(f"当前时间: {data.get('now')}") print(f"累计解码消息: {data.get('messages')}") print(f"当前跟踪飞机数: {len(aircraft_list)}") for ac in aircraft_list[:10]: flight = ac.get("flight", "").strip() squawk = ac.get("squawk", "0000") altitude = ac.get("alt_baro", None) speed = ac.get("gs", None) if flight or altitude or speed: print(f"呼号: {flight or '未知'}, 应答机编码: {squawk}, " f"高度: {altitude} ft, 地速: {speed} kt")这段代码作用很直接:通过 HTTP 接口拿全量飞机快照,再遍历每个飞机对象提取关键字段。相比 TCP 实时流,HTTP 方式的好处是拿到的是“当前时刻状态”,不需要自己维护连接状态和缓冲切分,适合写轮询型脚本。参数说明:urlopen 的 timeout 设为 10 秒,避免 dump1090 进程假死时脚本无限等待;航班呼号字段可能带前后空格,用 strip() 整理干净;alt_baro 是气压高度,单位英尺;gs 是地速,单位节。如果真的能看到飞机列表打出来,说明接收机、解码器、Python 读取三层已经全部打通,后面你想跟踪哪架、统计什么数据都从这一步开始。
4. 把裸报文变成可读情报:解析、过滤与地图标绘
4.1 从 JSON 字段到 Flight 对象:关键字段的取舍
3.3 节我们已经能从 HTTP 接口拿到原始 JSON,实际做分析时最好把散落的字段封装成 Flight 对象,避免后续代码里到处用字典索引且频繁判空。封装时有个核心技术决策:哪些字段要进对象,哪些字段不要。我的原则是“按分析场景最小化字段集”,因为多一个字段就多一份解析和兼容成本。下面实现包含基础身份信息(hex 地址、呼号、应答机编码)和位置速度信息(经纬度、气压高度、地速、航向)。
from dataclasses import dataclass from typing import Optional from datetime import datetime @dataclass class Flight: hex_ident: str # ICAO 24位地址 flight: str # 呼号,可能为空 squawk: str # 应答机编码 altitude: Optional[float] = None # 气压高度,单位英尺 latitude: Optional[float] = None # 纬度 longitude: Optional[float] = None # 经度 groundspeed: Optional[float] = None # 地速,单位节 track: Optional[float] = None # 航向角,0-360度 seen_at: Optional[datetime] = None # 数据接收时间 @classmethod def from_json(cls, item: dict) -> "Flight": """从 dump1090 的 aircraft 数组元素构建 Flight 对象""" return cls( hex_ident=item.get("hex", ""), flight=item.get("flight", "").strip(), squawk=item.get("squawk", "0000"), altitude=item.get("alt_baro"), latitude=item.get("lat"), longitude=item.get("lon"), groundspeed=item.get("gs"), track=item.get("track"), seen_at=datetime.fromtimestamp(item.get("seen", 0)) if item.get("seen") else None ) # 使用示例 sample = {"hex": "abc123", "flight": "ABC123 ", "squawk": "6345", "alt_baro": 35000, "lat": 39.897, "lon": 116.383, "gs": 480, "track": 270, "seen": 1700000000} f = Flight.from_json(sample) print(f"已解析航班: {f.flight} {f.hex_ident}")这里用 dataclass 定义 Flight 对象,利用类型注解标注每个字段的形态,大幅提升代码可读性和 IDE 补全体验。from_json 作为类方法实现从原始字典到对象的转换逻辑,方便在别处复用。不容忽视的是经纬度和高度字段的类型——dump1090 返回的 JSON 中这些数值可能是整数也可能是浮点数,全部用 Optional[float] 兜底,避免传进 None 时类型错误。seen_at 字段由 dump1090 提供的时间戳转换而来,不过注意这个时间是接收时刻,不是航空器发报时刻,两者可能差几百毫秒到几秒,做轨迹分析时要容忍这个偏差。
4.2 坐标换算:经纬度转平面坐标的实用公式
拿到经纬度后不能直接在地图上画点,特别是涉及轨迹显示和距离计算时,必须把 WGS84 坐标换算成平面坐标或投影坐标。有两个核心场景:一是短距离的欧氏距离计算,例如判断某个航班是否进入特定半径范围;二是航迹可视化,需要把经纬度映射到二维画布像素位置。
import math def haversine_meters(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """计算两点间的地面距离,返回米""" r_earth = 6371000 # 地球平均半径,单位米 phi1 = math.radians(lat1) phi2 = math.radians(lat2) dphi = math.radians(lat2 - lat1) dlambda = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2 c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) return r_earth * c def latlon_to_local(lat: float, lon: float, origin_lat: float, origin_lon: float) -> tuple: """ 将经纬度转换为以 origin 为原点的平面坐标(东向、北向),单位米。 适用于半径几百公里的区域,超出后精度下降。 """ x = haversine_meters(origin_lat, origin_lon, origin_lat, lon) y = haversine_meters(origin_lat, origin_lon, lat, origin_lon) # 根据目标点相对原点的方向确定正负号 if lon < origin_lon: x = -x if lat < origin_lat: y = -y return x, yhaversine 公式是球面两点距离的经典解法,误差在短距离上远小于地球曲率影响。这里用 6371000 米作为地球平均半径,实测与高精度专业算法在中低纬度误差可以控制在 1% 以内。latlon_to_local 函数把经纬度投影成以基点为中心的局部平面坐标,做法是两次调用 haversine 分别求东向偏差和北向偏差,再根据方向定正负。这个方法的适用范围建议在 300 公里半径内,超过了要考虑使用正式的 UTM 投影。实际使用中我发现有人直接拿经纬度差值乘以固定系数换算米数,这在低纬度和高纬度地区都会产生显著误差,而 haversine 实现简单且精度足够。
4.3 实时标绘:把飞机画上地图的轻量方案
当你把 Flight 对象和坐标换算都搞定,最直观的效果就是把飞机画到地图上。地图库选择上看你的部署环境:如果在浏览器里做可视化,开箱即用的是 Leaflet + OpenStreetMap 瓦片;如果只做本地桌面原型,用 Matplotlib 画底图加散点就够。重要的是不要一开始就引入重型三维地球库,那会掩盖你的数据质量问题。
import json import matplotlib.pyplot as plt from matplotlib.patches import Circle # 假设已经通过前面的方法拿到 flights 列表,每个元素是 Flight 对象 center_lat, center_lon = 39.9042, 116.4074 # 以某市中心为圆心 radius_km = 50 fig, ax = plt.subplots(figsize=(10, 10)) ax.set_aspect("equal") circle = Circle((0, 0), radius_km * 1000, fill=False, edgecolor="red", linestyle="--") ax.add_patch(circle) for flight in flights: if flight.latitude is None or flight.longitude is None: continue # 转换为以中心为原点的平面坐标 x, y = latlon_to_local(flight.latitude, flight.longitude, center_lat, center_lon) in_radius = haversine_meters(center_lat, center_lon, flight.latitude, flight.longitude) <= radius_km * 1000 # 距离中心半径内的点用红色标注,圈外用蓝色 color = "red" if in_radius else "blue" label = flight.flight if flight.flight else flight.hex_ident ax.scatter(x, y, s=50, color=color, alpha=0.7) ax.annotate(label, (x, y), textcoords="offset points", xytext=(5, 5), fontsize=8) ax.set_xlabel("东向 (米)") ax.set_ylabel("北向 (米)") ax.set_title(f"周边 {radius_km} 公里航班分布") plt.grid(alpha=0.3) plt.show()这段代码把每架飞机的位置从经纬度转成以雷达站为中心的平面坐标,然后按距离范围标成不同颜色。circle 的半径用了 radius_km * 1000 转成米,latlon_to_local 返回的单位正好也是米,两边口径一致。要注意的是 Matplotlib 的坐标轴比例,如果不强制 ax.set_aspect("equal"),地图会在纵向上被拉伸变形,导致距离看起来失真。若航班列表为空,可能的原因不是代码问题,而是当前天线周围确实没有飞机,或者 RTL-SDR 的增益太低接收不到信号,可以打开 dump1090 的 Web 页面确认一下消息量再回来调图。
5. 避坑与排查:收不到信号、字段为空、坐标偏了怎么办
5.1 现象:TCP 端口能连上,但 Python 迟迟收不到数据
连接成功但 recv 一直阻塞,这是最常见的翻车场景。原因往往是 dump1090 启动了 TCP 服务,但没有数据从 30003 端口流出——换句话说,dump1090 本身的接收链路可能就没通。排查步骤:先 curl 一下 HTTP 接口看 aircraft.json 是否非空,如果 HTTP 接口也没有数据,说明问题不在 TCP 而在射频接收层,检查天线是否接好、RTL-SDR 的灯是否亮起、USB 口是否供电不足。如果是树莓派外接硬盘再接 SDR,USB 供电不足的概率极高,换独立供电的 USB Hub 是常用解法。
5.2 现象:经纬度字段大量为空,但高度和速度正常
位置信息缺失但其他字段正常,大概率不是接收问题,而是 CPR 解码的奇偶配对不完整。ADS-B 的位置报文每对只发一个纬度或经度分量,需要连续两帧才能解出完整坐标。如果 dump1090 的 --fix 参数没开,弱信号下错帧率会明显升高,导致可配对的有效帧比例下降。解决方式是给 dump1090 加 --fix 参数,同时适当降低接收机增益——盲区错觉是增益越高越好,实际上在城市环境中高增益会引入更多互调干扰和底噪放大,信号质量不升反降。每台设备的天线位置和环境不同,增益最优值需要做一轮从低到高的扫描对比。
5.3 现象:坐标偏移或跳变,飞机位置在真实航线外的几十公里处
如果你确认数据源没问题,但某几个飞机的坐标在地图上明显跑偏,先别急着怪 Python 解析。这种单点偏移常用因是 dump1090 输出的经纬度来自 CPR 解算,奇偶帧配对错误时会产生“镜像点”或者跳变到航迹对称的另一侧。这个问题的根源在于接收机收到了干扰信号或报文本身有 CRC 漏检。处理办法:在 Python 侧做位置连续性校验——如果某架飞机在 2 秒内位置跳变了超过 1 公里,就标记为可疑数据,不参与轨迹拼接。下面是一个简易的过滤器:
class TrackFilter: def __init__(self, max_speed_kmh: float = 1500): self.max_speed_kmh = max_speed_kmh self.last_pos = {} def is_valid_move(self, icao: str, lat: float, lon: float, now: float) -> bool: """ 检查当前位置是否合理。 民用客机的正常地速一般不超过 1200 km/h,超过阈值的跳跃视为无效。 """ prev = self.last_pos.get(icao) if prev is None: self.last_pos[icao] = (lat, lon, now) return True prev_lat, prev_lon, prev_time = prev dt = now - prev_time if dt <= 0: return True distance = haversine_meters(prev_lat, prev_lon, lat, lon) speed = distance / (dt / 3600) if speed > self.max_speed_kmh * 1000: return False self.last_pos[icao] = (lat, lon, now) return True这里把相邻两个时间点之间的距离和时间差相除得到瞬时地速,如果超过设定的 1500 km/h 上限就认为数据跳变。飞机实际速度很少超过 1200 km/h,所以这个阈值留有空间。dt 为 0 时不判断,因为同一秒内收到两帧位置变化可能是正常的连续航迹点。这个过滤器的副作用是可能把真正的数据更新丢弃,但概率很低,优先级是保证轨迹平滑而非绝对实时。
5.4 现象:多台接收站数据打架,同一航班出现两个位置
如果你像很多人一样部署了多台接收机,覆盖不同的方向,同一个飞机的数据可能被两个站点同时上报,而它们的时间基准不同(每台机器的系统时钟有偏差),导致 Python 收到位置后频繁前后跳跃。常见做法是加一个以 hex 地址为键的缓存,每条数据进来时用时间戳做权重,只保留时间较新的状态。如果你用的是本地接收机加公共数据流的混合模式,公共流传输有额外延迟,不能直接把两边数据合并,否则坐标会在真实位置和延迟位置之间来回跳。我的选择是双源时只用一个主源,另一个作为离线后才看的备份。
5.5 现象:dump1090 启动后程序退出,没有报错信息
如果命令行一跑就退出且没有任何错误输出,这通常不是 dump1090 的问题,而是没有权限访问 USB 设备。RTL-SDR 设备默认属于 root 用户或特定用户组,普通用户无法直接打开设备节点。解决方式是执行以下命令把当前用户加入对应组,然后重新插拔设备:
sudo usermod -a -G plugdev $USER sudo usermod -a -G dialout $USER sudo udevadm control --reload-rules大部分发行版的 SDR 设备节点归属于 plugdev 组,也有部分归属于 dialout。把当前用户加入两个组没有坏处。执行完后重新登录或 newgrp 让权限生效。如果你用的是 Docker 部署 dump1090,还要额外加 --device 参数把 USB 设备映射进容器,否则容器内根本看不到设备。
6. 进阶:把数据存下来做离线回放,比实时看更有价值
实时看数据再炫也会审美疲劳,真正实用的玩法是把 ADS-B 数据落盘存成带时间戳的记录,之后可以离线调试算法、回溯某个航班轨迹、或者给图像处理 Demo 供数据。这个做法我现在几乎每次都会用:开一个后台进程持续接收 dump1090 的 TCP 流,写入压缩 JSONL 文件,按小时切片。这样就算当天没来得及分析,晚上回去也还有完整的数据做复盘。
import gzip import json import time from datetime import datetime def save_aircraft_snapshot(filepath: str, aircraft_list: list): """把指定时刻的飞机快照写入压缩文件""" snapshot = { "timestamp": time.time(), "iso_time": datetime.utcnow().isoformat(), "aircraft": aircraft_list } with gzip.open(filepath, "at", encoding="utf-8") as f: f.write(json.dumps(snapshot) + "\n") def collector_loop(interval_sec: int = 5): """每 5 秒做一次快照,按小时切换文件""" current_hour = None while True: ts = time.time() hour_str = datetime.fromtimestamp(ts).strftime("%Y%m%d%H") if hour_str != current_hour: current_hour = hour_str filepath = f"adsb_{current_hour}.jsonl.gz" print(f"切换到文件: {filepath}") # 这里读取 dump1090 的 aircraft.json,代码略 snapshot = fetch_aircraft_data() save_aircraft_snapshot(filepath, snapshot) time.sleep(interval_sec)这段代码是数据采集器的骨架,核心逻辑是按小时切换文件避免单个文件过大。gzip 压缩后一小时的数据大约几十到几百 KB,几十个小时也就几十 MB,完全可控。采集频率上 5 秒一次意味着航线轨迹上每个点间隔约几百米,对大多数分析场景足够了,如果你在意更精细的轨迹,可以压到 1 秒,但要注意 gzip 写入的 I/O 频率也同步增加。
离线回放的价值在于你可以快速遍历时间窗口内的所有快照,把某一架飞机的完整轨迹抽出来,复现当时的飞行路线。这在调试 2.3 节里的 TrackFilter 时特别实用——你知道某个时间点屏幕上有跳变,直接从回放数据中精确定位那一帧,观察前后的数据上下文,再调整过滤参数。如果是纯实时环境,你不会有机会反复重看同一段数据。
从那以后我每次搭建 ADS-B 接收链条,都会先把数据采集器跑起来再谈可视化,因为可视化代码改三遍、数据却没有留存的话,所有调试都是“盲改”。有了离线回放,调试变成了“打开历史数据 - 跑新算法 - 对比结果”的闭环,效率提升是质的。希望这篇拆解能帮你把 ADS-B 这套链路走通,少踩几个我已经替你踩过的坑。
本文还有配套的精品资源,点击获取