ADS-B Python实战:从RTL-SDR信号到航班数据解码与可视化
2026/9/8 13:07:50 网站建设 项目流程

简介:面向Python开发者的ADS-B消息处理工具包,专门用于解析广播式自动相关监视(ADS-B)报文数据。ADS-B是现代航空监视的重要技术之一,该工具为从事航空数据采集、空域监控、航迹分析或无人机监管的开发者提供了解析消息的基础能力,整体代码轻量、方便集成。

压缩包内共49个文件,以Python源码为主(23个.py文件),包含测试用例、示例脚本、消息日志等运行素材;另有8个rst文档用于技术说明、5个txt文本、2个YAML配置和Makefile构建文件等,整体仅约55KB,便于通读源码和二次开发。已有1013人下载学习。

通过该工具包,开发者可以梳理ADS-B消息的解析流程,理解报文处理常见思路,同时参考项目的目录结构、测试组织方式与CI配置,快速上手自己的Python工具模块。资源中还附带消息日志与示例数据,可用于实验和效果验证,是一份性价比很高的轻量学习参考。 直接开工,写一篇从零看懂、能照着跑的ADS-B Python工具博文。

1. 认识ADS-B:天空中的数字广播,以及Python为什么要掺一脚

你抬头看到一架飞机飞过,想知道它从哪来、飞多高、往哪去,这在十年前大概要查机场大屏或者等落地广播。现在不用了,因为绝大多数民航客机都在用一套叫做ADS-B(Automatic Dependent Surveillance-Broadcast,广播式自动相关监视)的系统,主动向外广播自己的位置、高度、速度、航班号,这些信息就在我们头顶的空气里,用1090MHz的频率往外发。

ADS-B的大致机制可以这样理解:飞机上的GPS接收机算出了自己的位置和速度,然后通过应答机(Transponder)把这些信息打包,周期性地广播出来。地面接收站(比如民航局的监视雷达网)收到后就能实时掌握空域状态。但问题在于,这套广播是开放接口、明文传输的,只要有一台能接收1090MHz信号的设备,再加一套解码工具,任何人都能还原出整个天空的交通画面。

普通人能拿到什么?硬件方面有相当成熟的RTL-SDR方案,一根USB电视棒改装的接收器,价格几十到几百元。软件方面,过去大家习惯用dump1090这类C写的工具,把原始信号解码成网页JSON。可一旦你想在脚本里分析航班轨迹、做密度统计、甚至把数据接到自己的可视化平台,C工具的封闭性就让人觉得别扭。这时候Python的价值就出来了:数据分析生态齐全,开发迭代快,而且社区里已经有相当成熟的ADS-B解码库,不需要自己从比特级折腾起。

用Python处理ADS-B解决的核心问题,是把“你能收到飞机信号”升级成“你能理解飞机信号”。如果不做解码,RTL-SDR给你的只是带I/Q采样的原始数据或者一串十六进制帧;只有解码之后,你才能说出“这架飞机高度10100米,地速850公里每小时,正在向北偏东飞行”。本文面向的是:

  • 刚接触ADS-B、想用Python做课程设计或个人项目的学生或爱好者;
  • 有一定Python基础但没碰过无线电数据处理的开发者;
  • 已经在用dump1090但觉得数据格式不够灵活、想自己写解析逻辑的玩家。

接下来,我会从底层帧格式讲起,到环境搭建、数据接入、解码实现、可视化输出完整走一遍,最后把常见坑按“出现原因+排查步骤+解决动作”的方式给你列清楚,保证你看完能独立跑通一个完整项目。

2. 从零开始:整体设计思路与方案选型

2.1 为什么是“解码库+业务脚本”而不是搞一个重型框架

ADS-B工具的核心能力就两件事:拿到原始数据,把原始数据变成结构化信息。很多人刚开始会犯一个错——上来就封装类、搞模块化架构、写一堆抽象接口。实际上这类工具的数据流是极简的,我建议的架构是:

采集层(RTL-SDR / 网口数据源) → 解码层(pyModeS / 自研解码) → 业务层(数据落库 / 前端展示 / 统计分析)

解码层是核心,业务层完全看你想做什么。根本没有必要引入数据库抽象层、消息队列这些重型组件,除非你要做全国范围的接收站组网。个人项目和中小规模实验,一个脚本文件加一个配置文件就够了。

方案选型上,成熟社区方案有三个方向:

  • 使用 pyModeS 这类现成库做纯Python解码;
  • 调用 dump1090 的 --json 输出,再用Python消费JSON流(本质是用外部工具解码);
  • 完全自研解码器,逐位解析二进制帧。

我的建议:除非你有教学或研究目的,否则不要选第三个。ADS-B解码头两步(位同步、CRC校验)看起来简单,但报文类型多到能写出一本手册,纯手写很容易在细节上翻车。第二个方案部署最快,适合时间紧迫、只想出可视化的场景。第一个方案才是“Python工具”的正解——保留了解码过程的透明性,同时不用从零造轮子。

从数据链路选择上说,如果你已经有RTL-SDR设备,推荐用rtlsdr库直接读取I/Q采样,然后交给解调函数处理;如果你只有一台装有dump1090的树莓派在远处,那就用HTTP流或者TCP流拉取已经解出的Message数据。两种接入方式在后面的章节都会有代码示例。

2.2 关键设计:解码数据的标准形态与输出格式

ADS-B每条报文在解码后应该统一成一个独立的数据对象。以我常用的数据结构为例,一条数据记录至少包含:

字段含义示例
ICAO飞机唯一地址码780D2C
callsign航班号CCA1837
altitude气压高度(英尺或米)39000 ft
lat/lon经纬度(可能来自CPR解码)40.02, 116.23
ground_speed地速(节)455 kt
track航向角(度)183
timestamp接收到报文的时间2024-06-01 12:00:00.123

这个字典结构的设计有个隐性好处:后面无论是存CSV、入SQLite、还是发到InfluxDB做时序展示,都只是一行JSON序列化的问题。避免为每个输出目标单独写一套字段映射,是这类小工具保持简洁的关键。

还有一点值得从一开始就明确:ADS-B数据的实时性和完整性有波动。你在脚本里必须对“某架飞机某一条消息缺失高度”这类场景有容忍度,不能因为它没上报就中断流程或抛异常。这个思路会体现在后面的解码函数实现里——每一层都用try/except兜住,宁可返回None,也不要让一个坏消息干掉整批处理。

3. Python环境准备:从安装到依赖库选型

3.1 环境安装:Python版本怎么选、虚拟环境怎么建

用到这套工具,Python 3.8+ 就够了。如果你用的是Windows,去python.org下载安装包时记得勾选“Add Python to PATH”;macOS用户建议直接用Homebrew安装,终端执行:

brew install python@3.11

Linux发行版的包管理器通常也自带Python,但版本可能偏旧,动手之前先核对一下:

python3 --version

强烈建议使用虚拟环境,把项目依赖锁定在当前目录,不污染全局环境。在项目文件夹里执行:

python3 -m venv venv source venv/bin/activate # Windows下执行 venv\Scripts\activate

激活后可以看到终端前缀变成(venv),后面pip装的包就都隔离在这个环境里了。

3.2 核心依赖:pyModeS、numpy、SQLite标准库的取舍

解码层依赖推荐pyModeS,这是目前Python生态里ADS-B解码最成熟的库,支持Mode-S报文的基础校验、ICAO提取、BDS解译、CPR位置解码。安装非常省事:

pip install pyModeS

如果你要自己处理I/Q数据(不依赖dump1090解调),还需要numpy处理数组、rtlsdr读取USB设备:

pip install numpy rtl-sdr

注意rtl-sdr的pip包名是rtl-sdr,代码里import时写的是rtlsdr。这个小细节当年让我卡了一会儿,习惯性pip install rtlsdr,会告诉你找不到包。

数据落库阶段我偏爱SQLite,因为它是Python内置标准库sqlite3,零安装、单文件、查询方便,存几万条航班记录毫无压力。可视化环节如果需要出地图轨迹,可以按需加folium,这是个非常轻的地图可视化库,之后在实战环节会看到具体用法。

先装好上面这些,环境就齐了。

4. 实战:完整接入1090MHz ADS-B数据流

4.1 方式一:直接读RTL-SDR,从I/Q信号起步

这种方式最能体会到“无线电数据”到“航班信息”的完整链路。代码逻辑不复杂,用rtlsdr读出一段I/Q采样,然后交给解调函数做幅度判决,提取脉冲位置,再据此还原二进制报文。

from rtlsdr import RtlSdr sdr = RtlSdr() sdr.sample_rate = 2_000_000 # 采样率2MHz,1090MHz信号带宽约1MHz,够用 sdr.center_freq = 1_090_000_000 # 1090MHz,ADS-B专用频率 sdr.gain = 'auto' # 自动增益,室内测试首选 samples = sdr.read_samples(256 * 1024) # 读取256K个采样点 print(samples[:10])

得到的复数数组需要经过解调:取实部或取模,用阈值判断高低电平变化,7090采样点对应一个完整脉冲序列。完整解调函数有几个细节要考虑(脉冲宽度校验、前导脉冲检测),这部分我会在第5章拆解。

注意:树莓派或普通笔记本USB口的供电能力有限,RTL-SDR在采样率高时容易丢失采样,表现为“Message碎片化”,解码成功率骤降。如果遇到这种情况,先试试把采样率降到1MHz或者用带供电的USB HUB。

4.2 方式二:对接已有接收机,拉JSON流

如果你已经有树莓派跑着dump1090,或者从公网上找到了可靠的ADS-B聚合数据源(比如某些航空社区开放的API),就可以跳过射频解调,直接消费数据流。以dump1090的--net输出为例,它默认在端口30003暴露原始的逗号分隔报文流,同时会在8080端口提供JSON接口。

用Python消费30003端口的原始报文,核心代码就一段socket读取:

import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(('192.168.1.100', 30003)) # 换成你接收机的地址 buffer = b'' while True: data = sock.recv(4096) if not data: break buffer += data while b'\n' in buffer: line, buffer = buffer.split(b'\n', 1) msg = line.decode('latin-1') # 这里msg形如: MSG,3,111,11111,78D2C5,111111,2024/06/01,12:00:00,... process_message(msg)

这种方式的典型好处是省掉了射频硬件调试,也避免了Python解码I/Q数据对CPU的消耗。代价是你依赖外部接收机的解调质量——如果它在楼顶接收效果差,你拿到手的报文也会有严重漏报。

4.3 数据格式初识:读懂一条裸报文

在开始解码之前,建议先人工看一条报文。即便走RTL-SDR路线,解调完成后得到的也是一个十六进制字符串,类似:

8D780D2C990C6C6A28943F8010B6

这个字符串分成两部分含义:前缀8D说明这是DF17类型的ADS-B报文(Mode-S Extended Squitter),其余部分包含ICAO地址、ME字段和CRC校验。你在业务代码里拿到这个字符串后,首先要做的是校验CRC,丢弃损坏数据,然后才进入ME字段解析。

import pyModeS as pms raw = '8D780D2C990C6C6A28943F8010B6' print(pms.icao(raw)) # 输出 ICAO 地址 print(pms.crc(raw)) # 返回0说明数据有效 typecode = pms.adsb.typecode(raw) # 数据类型编号 print(typecode)

typecode是解码分流的关键钥匙:1~4是识别信息,5~8是地面位置,9~18是空中位置,19是速度信息,20~22是空中位置(带高度更高精度)。看到typecode后,你就知道该调用哪类解析函数了。

5. 核心代码实现:解码航班位置与状态数据

5.1 解码CEP报文:识别航班号和ICAO地址

在民航监视术语里,CEP(Common Extended Protocol)报文承载航班识别信息。报文开头已经包含ICAO地址,pyModeS一条调用即可完成航班号提取:

import pyModeS as pms raw = '8D780D2C990C6C6A28943F8010B6' icao = pms.icao(raw) # 780D2C callsign = pms.adsb.callsign(raw) # 返回 'CCA1837' 或类似 print(f"ICAO: {icao}, 航班号: {callsign}")

有个细节容易忽略:航班号在ADS-B里是8个字符,比如UAL123会被补成UAL123(尾部空格)。存储到数据库之前,最好做一次.strip(),否则后面做搜索匹配时会踩坑。

5.2 空中位置解码:CPR算法的关键步骤

位置数据是ADS-B最惊艳的部分。飞机广播的经纬度并不是直接数值,而是用一种叫CPR(Compact Position Reporting,紧凑位置报告)的编码方式传输。核心思路是:把地球用2的17次方格网划分,广播的是当前格网内的相对位置和格网索引,结合奇偶帧两次报文的协同,才能算出绝对经纬度。

import pyModeS as pms # 同一架飞机的两条连续报文,一条使用偶数帧,一条使用奇数帧 msg_even = '8D780D2C990C6C6A28943F8010B6' msg_odd = '8D780D2C990C6C6A28943F8010B7' # 示例延后一位 position = pms.adsb.position(msg_even, msg_odd, t_even=0, t_odd=1) print(position) # (lat, lon) 元组

CPR解码有个至关重要的前提:奇偶两条报文的观测时间差不能超过10秒,否则地球自转和飞机位移带来的误差会大到完全不可用。实际项目中,你应该为每架飞机维护它的最近一条odd帧和最近一条even帧,并记录时间戳。收到新报文后,先看是否与已有帧构成奇偶配对,配对成功再解位置。

如果只有单条报文,也可以调用pms.adsb.position_with_ref(),传入一个接近真实位置的参考点(比如接收站自己的经纬度),然后解码出本地坐标系下的相对位置。单条报文的精度不如奇偶配对,但做近距离监控完全够用。

5.3 速度与高度抽取:字段级别的位运算

速度信息来自TC=19的报文,包含地速、航向角、垂直速度等字段。pyModeS封装很完善:

speed_info = pms.adsb.velocity(raw) # {'speed': 455, 'track': 183.2, 'vertical_rate': -64, 'speed_type': 'GS'}

高度信息在TC=9~18的报文里,区分气压高度和几何高度:

altitude = pms.adsb.altitude(raw) print(altitude)

这里有个需要澄清的点:ADS-B广播的高度默认是英尺。如果你要做民航业务内的数据展示,换算成米时除以3.2808即可。但如果你要把数据放进飞行轨迹分析,建议保留英尺原始值,避免精度损失。

5.4 把解码逻辑封装成消息处理器

最终你的数据处理入口应是一个分发函数,根据typecode决定走哪条解析逻辑,然后把结果合并成一份统一字典:

def process_adsb_message(msg_hex: str, timestamp: datetime): result = { 'icao': pms.icao(msg_hex), 'timestamp': timestamp.isoformat(), 'raw': msg_hex } tc = pms.adsb.typecode(msg_hex) if tc in range(1, 5): result['callsign'] = pms.adsb.callsign(msg_hex).strip() elif tc in range(9, 19): result['altitude'] = pms.adsb.altitude(msg_hex) result['position'] = pms.adsb.position_with_ref( msg_hex, ref_lat=40.0, ref_lon=116.0 ) elif tc == 19: vel = pms.adsb.velocity(msg_hex) result['ground_speed'] = vel.get('speed') result['track'] = vel.get('track') result['vertical_rate'] = vel.get('vertical_rate') return result

注意这里用了position_with_ref,适配“单帧位置+参考点”的场景;如果你要追求高精度,就得另写一个维护奇偶帧缓冲区的类,核心逻辑都一样,差别在于传参。

6. 进阶:数据落库与可视化,把工具变成产品

6.1 用SQLite持续记录航班动态数据

一旦解码跑通,数据落库是下一个马上能见效益的优化。这里分享我常用的一个轨迹表结构:

CREATE TABLE IF NOT EXISTS flights ( id INTEGER PRIMARY KEY AUTOINCREMENT, icao TEXT, callsign TEXT, altitude_ft INTEGER, lat REAL, lon REAL, ground_speed_kt REAL, track_deg REAL, ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_flights_icao ON flights(icao); CREATE INDEX IF NOT EXISTS idx_flights_ts ON flights(ts);

把第5章process_adsb_message()返回的字典直接插入这张表:

import sqlite3 conn = sqlite3.connect('adsb.db') def save_to_db(data): conn.execute( "INSERT INTO flights (icao, callsign, altitude_ft, lat, lon, ground_speed_kt, track_deg) " "VALUES (?, ?, ?, ?, ?, ?, ?)", ( data.get('icao'), data.get('callsign'), data.get('altitude'), data.get('position', (None, None))[0], data.get('position', (None, None))[1], data.get('ground_speed'), data.get('track'), ) ) conn.commit()

踩坑提醒:不要每条报文都commit()。Python的sqlite3默认每次commit都会触发磁盘写入,高频数据流下会让I/O成为瓶颈。正确做法是攒一批数据,比如每当积累到200条或者每隔2秒,统一commit一次,性能能提升一个数量级。

6.2 地图可视化:基于当前接收范围的实时态势

拿到了数据、存了库,最后一步是把它变成一眼能看懂的态势图。首推方案是folium,它基于Leaflet地图,搭配Vue前端或者纯HTML都能快速出效果。

import folium m = folium.Map(location=[40.0, 116.0], zoom_start=8) for row in conn.execute( "SELECT callsign, lat, lon, altitude_ft, ground_speed_kt FROM flights WHERE lat IS NOT NULL ORDER BY ts DESC LIMIT 100" ): cs = row[0] or 'UNKNOWN' folium.CircleMarker( location=[row[1], row[2]], radius=5, popup=f"{cs} 高度:{row[3]}ft 速度:{row[4]}kt", color='crimson', fill=True, ).add_to(m) m.save('flights.html')

有了这个HTML文件,直接用浏览器打开就能看到当前接收范围内所有飞机的分布。如果想做成实时刷新版,可以加一个5秒定时器重新加载数据源,或者把数据接口暴露为FastAPI端点,前端轮询刷新。

6.3 性能优化方向:多线程接收、批量入库、长时间采集

ADS-B数据是典型的高频小报文流。单线程循环做“接收→解码→入库”看起来没毛病,实际上在数据量大的时候会互相卡。我的建议是拆两条线程:

  • 接收线程:持续从RTL-SDR/socket读取数据,解码完放进queue.Queue
  • 消费线程:从队列取数据,批量写SQLite、更新可见飞机列表。
import queue import threading msg_queue = queue.Queue(maxsize=1000) def consumer(): while True: item = msg_queue.get() if item is None: break save_to_db(item) threading.Thread(target=consumer, daemon=True).start() # 接收循环中 try: dec = process_adsb_message(raw_hex, datetime.now()) if dec: msg_queue.put(dec) except Exception as e: # 坏报文直接跳过,别影响主循环 pass

队列长度上限很重要,防止消费端变慢时内存无限增长。这套结构在树莓派3B+上跑满2MHz采样率,CPU占用也稳定在20%以下。

7. 常见问题与排查技巧实录

7.1 收不到信号 / 解码率为零

排查步骤:

  1. 先确认天线环境。室内阳台和窗外接收效果相差巨大,1090MHz信号对遮挡和金属框架非常敏感;
  2. 确认center_freq = 1090MHzsample_rate设置正确。常见的错是中心频率写成了1090kHz;
  3. 检查增益。RTL-SDR增益设成auto有时会偏低,手动调到40dB左右通常能大幅改善;
  4. pms.crc(raw)对原始报文做校验,如果大量报文CRC都失败,说明解调出来的数据误码率高,大概率是信号太弱或采样率过低。

7.2 pyModeS的adsb模块提示typecode无法识别

大概率是传入的报文并非ADS-B类型(比如是Mode-S短应答或普通长应答)。做数据处理时,先判断首两位是否为8D(DF17),DF18也携带ADS-B信息,但格式略有差异,可以先统一过滤掉。

7.3 位置解码跳变严重

原因通常是CPR奇偶帧配对不当,比如把不同时刻、位置相距甚远的帧硬拼在一起来解码。解决方式:为每架飞机维护最近20秒内的奇偶帧缓冲区,只有都在时间窗内时才解码;否则回退使用position_with_ref

7.4 CPU占用过高,树莓派发热严重

优先检查是不是每条报文都触发了数据库commit()。改批量提交后,CPU占用通常会明显下降。其次可以考虑降低采样率到1.5MHz甚至1MHz,ADS-B信号实际占用约1MHz带宽,2MHz是冗余而非必需。

7.5 时间戳不准确,轨迹时间线错乱

RTL-SDR本身没有可靠的时间戳机制,read_samples()返回数据时的时间已经是接收后的一段时间。如果你要做精确到秒级的轨迹分析,建议给每个解码结果打上“处理时刻”而非“接收时刻”,同时在仪表盘上标注精度范围。

8. 写在最后:这个项目还能怎么扩展

做完一遍之后,我最大的体会是:ADS-B + Python这个组合,真正有意思的地方不在解码本身,而在于解码之后你能做的分析。比如把一整天的数据存下来,就能统计某条航路的航班密度高峰时段;接入多个接收站的数据,可以做简单的多点定位增强;把历史和实时数据叠加,甚至能做一个迷你的空域态势回放系统。

最后分享一个调试小技巧:跑这类实时数据项目,刚开始不要急着接真实数据源。先用dump1090 --net --write-json保存一段几分钟的原始数据,或者直接下载一份公开的ADS-B样例数据集,在文件上调试解码逻辑,等确认解析完全正确再接硬件。这个习惯能帮你省掉至少半天的调试时间。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询