简介:这份资源是面向Python开发者的ADS-B消息处理工具包,适合对航空监视数据解析、飞行器追踪感兴趣的初学者与中级开发者,用于在Python项目中快速接入并处理ADS-B报文。压缩包共49个文件,约55KB,以23个py源码文件为核心,辅以8个rst文档、5个txt说明及yml、makefile等构建配置,另含kml示例与测试脚本,目录涵盖src、tests、docs、examples等模块,结构清晰便于按需查阅。目前已有1017人学习下载,具备一定参考基础。读者可获得可直接pip安装的adsb包源码、API级别文档与基础用例,借助tests中的消息解析、客户端服务端及JSON工具测试理解报文处理流程,并通过examples与kml文件快速验证输出效果,为后续扩展ADS-B数据解析功能提供可复用的起点。
1. 从一条 1090MHz 报文说起:这个 Python 工具到底能干什么
如果你手上有 RTL-SDR 或者别的 SDR 硬件,想抓飞机广播的 ADS-B 报文,第一道坎往往不是硬件,而是“抓下来的一堆十六进制字符串怎么变成经纬度、高度、航班号”。adsb这个 Python 工具包就是干这件事的:它把 1090MHz 链路上收到的 Mode S / ADS-B 原始报文做解码,输出结构化的飞行器状态。我第一次接触它是在一个模拟项目里,需要把离线录制的 IQ 采样转成轨迹点,当时试了几个方案,最后发现这个包的接口最贴近“拿来就能跑”的预期——不用自己啃 DO-260B 文档去拼 CPR 编码。
它适合两类人:一类是做无线电信号处理、想快速验证解码链路是否通的工程师;另一类是做数据分析、手里已经有报文文本、只想批量解析出结构化字段的开发者。不适合指望它直接驱动硬件收信号的人——它管的是解码,不是采集。下面按“装起来跑通 → 报文怎么解 → 参数怎么调 → 坑在哪”的顺序拆。
2. 环境搭起来:依赖、安装与第一条报文解码
2.1 为什么选 Python 而不是现成的 C 工具
市面上解码 ADS-B 的成熟工具有不少是 C 写的,性能好但改起来费劲。选 Python 版本的核心动机是“可嵌入”:你要把解码结果接进 pandas 做轨迹分析、接进 Flask 做个简易监控页、或者塞进一个模拟项目的数据管道里,Python 的胶水能力省掉大量序列化开销。adsb这类包通常只依赖标准库加少量数值计算库,安装体积小,在树莓派这类边缘设备上也能跑。
常见做法是建独立虚拟环境,避免和系统里的 numpy 版本打架:
python -m venv adsb-env source adsb-env/bin/activate # Windows 用 adsb-env\Scripts\activate pip install adsb装完先确认版本和模块入口,别急着上真实数据:
import adsb print(adsb.__version__) # 确认装的是预期版本 print([n for n in dir(adsb) if not n.startswith('_')]) # 看暴露了哪些接口这一步的意义在于:不同小版本的模块组织方式可能不同,先看清顶层命名空间里有什么,比照着过时示例硬套要省时间。如果dir出来的名字和你手头教程对不上,八成是版本差异,先查包内__init__的导出列表。
2.2 用一条已知报文验证解码链路
在接真实硬件之前,最稳的验证方式是用一条公开的样例报文跑通全流程。ADS-B 的扩展电文里,DF17 是最常见的(位置、呼号、速度都走它)。下面这段演示把十六进制字符串转成二进制、再交给解码函数:
from adsb import decoder # 一条 DF17 扩展电文的十六进制表示(样例,用于验证链路) raw_hex = "8D40621D58C382D690C8AC2863A7" # 十六进制转字节,再转成位串,具体入口名以实际包为准 msg_bytes = bytes.fromhex(raw_hex) result = decoder.decode(msg_bytes) print(result) # 看返回的是 dict 还是对象 print(type(result))逻辑说明:bytes.fromhex负责把文本报文还原成二进制,这是所有解码器的统一入口;decode内部会先判 DF(下行格式)字段,再按 TC(类型码)分派到位置、识别、速度等子解码器。参数上唯一要注意的是输入必须是偶数长度的十六进制串,奇数长度会直接抛ValueError——这是新手第一个翻车点,后面避坑章会细说。
跑通之后,返回结构里通常能看到icao(24 位地址)、callsign、altitude、latitude、longitude这些字段。如果位置字段是None,别慌,单条报文解不出绝对位置是正常的,CPR 解码需要奇偶帧配对,下一章讲。
3. 报文解码的核心:CPR 配对、字段映射与批量处理
3.1 CPR 压缩位置为什么要成对解码
ADS-B 为了省带宽,位置不是直接发经纬度,而是发“压缩位置报告”(CPR),分奇帧和偶帧两种。单帧只能确定飞机在一个网格里的相对位置,必须一奇一偶配对才能算出全球唯一解。这是整个解码链路里最容易让人困惑的地方——很多人拿到单条报文解出None就以为工具坏了。
配对逻辑的要点是:两帧的 ICAO 地址必须相同,且时间间隔不能太大(常见实现里超过 10 秒就丢弃配对)。下面是一个简化的配对流程:
from adsb import decoder # 模拟缓存:按 ICAO 存最近一帧奇/偶报文 cache = {} def feed(raw_hex): msg = bytes.fromhex(raw_hex) parsed = decoder.decode(msg) icao = parsed.get("icao") if not icao: return None # 判断奇偶帧,字段名以实际返回为准 is_odd = parsed.get("odd_flag") cache.setdefault(icao, {})["odd" if is_odd else "even"] = parsed pair = cache[icao] if "odd" in pair and "even" in pair: # 两帧齐了,做全局解码 return decoder.decode_position(pair["even"], pair["odd"]) return None逻辑说明:feed每喂一条报文就更新缓存,凑齐奇偶后调用decode_position。参数上要盯住两点——odd_flag的取值语义(有的版本是布尔,有的是 0/1),以及配对的时间窗。如果你们的场景是离线回放,时间戳来自文件而非系统时钟,配对窗口要按报文时间算,不能按处理时间算,否则回放速度一快就全配不上。
3.2 字段映射:从原始报文到业务字段
解出来的字典字段名各版本略有差异,落地到业务表之前建议先做一层显式映射,别在代码里到处写parsed["alt"]这种硬编码。常见字段对照如下:
| 原始字段(常见名) | 含义 | 单位/取值 | 注意事项 |
|---|---|---|---|
icao | 24 位飞机地址 | 十六进制串 | 全局唯一,做主键 |
callsign | 航班呼号 | 字符串 | 常带尾部空格,需 strip |
altitude | 气压高度 | 英尺 | 可能为None(地面或无效) |
latitude/longitude | 经纬度 | 度 | 需 CPR 配对后才有值 |
ground_speed | 地速 | 节 | 速度类报文才有 |
track | 航迹角 | 度 | 0~360 |
vertical_rate | 垂直速率 | 英尺/分 | 正负号表示爬升/下降 |
这张表的价值在于:当你把解码结果写进数据库或 CSV 时,单位不统一是最隐蔽的坑。高度是英尺、速度是节,如果下游分析默认按米和千米每小时处理,轨迹图会离谱到没法看。我一般会在映射层直接做单位换算并加后缀(如altitude_ft),让字段名自带单位,省得后人猜。
3.3 批量处理离线报文文件
真实场景里更常见的是“手里有一个几百 MB 的报文文本,每行一条”,而不是实时流。批量处理的骨架是逐行读、逐条解、配对、落盘:
from adsb import decoder def process_file(path, out_path): cache = {} with open(path, "r", encoding="utf-8") as fin, \ open(out_path, "w", encoding="utf-8") as fout: for line in fin: raw = line.strip().lstrip("*").rstrip(";") # 去掉常见包裹符 if not raw: continue try: parsed = decoder.decode(bytes.fromhex(raw)) except ValueError: continue # 坏行跳过,别让整批挂掉 icao = parsed.get("icao") if not icao: continue slot = cache.setdefault(icao, {}) slot["odd" if parsed.get("odd_flag") else "even"] = parsed if "odd" in slot and "even" in slot: pos = decoder.decode_position(slot["even"], slot["odd"]) if pos: fout.write(f"{icao},{pos['latitude']},{pos['longitude']}\n")逻辑说明:lstrip("*").rstrip(";")处理的是很多抓包工具给报文加的包裹符,不去掉会导致fromhex直接失败;try/except ValueError是批量处理的后悔药,单条坏数据不该中断整批。参数上,输出只保留 ICAO 和经纬度是简化版,实际落地建议把时间戳、高度、呼号一起写出去,方便后续按时间排序和去重。注意缓存cache在长文件里会持续增长,如果 ICAO 数量极大,要加个过期清理,否则内存会慢慢涨上去。
4. 参数与边界:采样率、时间窗和精度取舍
4.1 解码前的信号层参数
虽然adsb管的是报文解码,但如果你是从 IQ 采样一路做下来,采样率和前级滤波会直接决定解码成功率。1090MHz 的 ADS-B 脉冲是 0.5 微秒量级,常见做法是采样率不低于 2MS/s,否则脉冲边沿糊掉, preamble 检测就过不去。这一层不在 Python 包里,但它是“为什么我解不出来”的头号嫌疑。
实操上我会先用一段已知良好的录制文件验证解码器本身没问题,再换自己的采集数据。如果已知文件能解、自己的解不出,问题一定在采集侧(增益太低、中心频率偏了、天线不匹配),而不是解码代码。这个二分法能省掉大量瞎调参数的时间。
4.2 配对时间窗怎么定
前面提到配对窗口常见是 10 秒,但这个值不是死的。高密度空域里飞机多、报文密,窗口可以收紧到 5 秒减少错配;低密度或丢包严重的场景,放宽到 15 秒能救回一些配对。判断依据是:如果你发现解出的位置跳变剧烈、轨迹出现瞬移,多半是奇偶帧配错了对象,这时候收紧窗口比调别的参数都管用。
4.3 精度与去重的取舍
同一架飞机在几秒内会发多条位置报文,直接落盘会产生大量重复点。常见做法是按(icao, 时间戳秒)去重,或者按最小位移阈值过滤(位移小于几十米就丢弃)。但阈值设太大,低速滑行的地面目标会被抹掉;设太小,悬停或盘旋的轨迹又会抖。我一般先不去重,把原始点画出来看分布,再决定阈值——先看数据再定参数,比拍脑袋强。
5. 避坑与排查:那些让我返工的问题
5.1 现象:fromhex报 “non-hexadecimal number found”
原因:报文里混了*、;、空格或换行,或者长度是奇数。很多抓包工具输出带包裹符,直接喂给bytes.fromhex必炸。 解决:入库前统一清洗,strip()后去掉首尾非十六进制字符,并断言长度为偶数。批量脚本里务必用try/except包住单条解码,别让一行脏数据毁掉整批。
5.2 现象:位置字段永远是None
原因:只喂了单帧,或者奇偶帧的 ICAO 对不上、时间差超窗。CPR 全局解码的前提是同一架飞机的一奇一偶。 解决:确认缓存按 ICAO 分组,检查odd_flag解析是否正确(有的版本字段名不同),并核对两帧时间戳差值。离线回放时尤其注意用报文时间而非处理时间做窗口判断。
5.3 现象:解出的经纬度落在海里或完全离谱
原因:奇偶帧配错,或者把不同飞机的帧混进同一个缓存槽。也可能是 CPR 的 NL(纬度分区)计算在极区附近有边界情况。 解决:配对前强校验 ICAO 相等;对高纬度数据单独抽查几个点人工核对;如果轨迹整体偏移一个固定量,检查是不是经纬度顺序写反了(lat/lon互换是经典翻车)。
5.4 现象:批量跑一半内存爆掉
原因:缓存字典只增不减,长文件里 ICAO 数量累积。 解决:给缓存加时间戳,定期清理超过配对窗口的条目;或者按时间分块处理,每块结束清空缓存。这个坑在短样例上永远暴露不出来,一上真实数据就现形。
5.5 现象:呼号带一堆尾部空格或乱码
原因:呼号字段是定长编码,不足位补空格;字符集映射不对时会出现非预期字符。 解决:读出后统一strip(),并对字符集做白名单过滤。别小看这个,呼号带空格会导致按呼号聚合时同一架飞机被拆成多条记录。
6. 进阶:把解码结果接进轨迹分析管道
跑通单条和批量之后,真正体现价值的是把解码输出接进分析链路。我常用的做法是解码层只负责“报文 → 结构化记录”,落成一张宽表,后面所有分析都基于这张表,避免每次分析都重跑解码。宽表字段至少包含:icao、timestamp、lat、lon、altitude_ft、ground_speed_kt、track_deg、vertical_rate_fpm。
验证解码质量有个简单办法:对同一架飞机按时间排序,算相邻点的位移速度,和报文里的ground_speed对比。两者量级差太多,说明配对或时间戳有问题。这个自检我每次接新数据源都会跑一遍,比肉眼盯轨迹图靠谱。
import math def haversine_m(lat1, lon1, lat2, lon2): R = 6371000.0 p1, p2 = math.radians(lat1), math.radians(lat2) dp = math.radians(lat2 - lat1) dl = math.radians(lon2 - lon1) a = math.sin(dp/2)**2 + math.cos(p1)*math.cos(p2)*math.sin(dl/2)**2 return 2 * R * math.asin(math.sqrt(a)) def check_speed(points): # points: [(t_sec, lat, lon, gs_kt), ...] 已按时间排序 for (t1, la1, lo1, gs), (t2, la2, lo2, _) in zip(points, points[1:]): dt = t2 - t1 if dt <= 0: continue v = haversine_m(la1, lo1, la2, lo2) / dt * 1.94384 # m/s 转节 if abs(v - gs) > 100: # 偏差超过 100 节就标记 print(f"可疑点 t={t2} 计算速度={v:.0f}kt 报文速度={gs}kt")逻辑说明:haversine_m算球面距离,check_speed用相邻点位移除以时间差得到推算速度,再和报文地速比对。参数100节是经验阈值,用来容忍时间戳抖动和位置量化误差;如果你的数据时间精度高,可以收紧到 50 节。这个自检能抓出大部分配对错误和单位混用问题。
从那以后我每次接新的报文数据源,都强制先跑一遍速度自检再往下做分析,宁可多花十分钟,也不要在轨迹图上对着一条瞬移的线猜半天。希望帮到你。
本文还有配套的精品资源,点击获取