ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

AIS数据解析全流程:从串口驱动到船舶轨迹可视化

AIS数据解析全流程:从串口驱动到船舶轨迹可视化

简介:本资源是一套面向嵌入式开发与海洋信息化方向初学者的AIS系统实践入门材料,聚焦自动识别系统的核心技术链:协议理解、信号解码、设备驱动开发与数据解析应用。资源共3个文件,含2个文本说明文档(涵盖ITU-R M.1371报文结构、VHF双信道传输机制、MMSI编码规则等关键知识点)和1个C++实现的AIS输出解析示例(output.cpp),代码注释清晰,覆盖二进制位解析、坐标转换与基础错误校验逻辑;压缩包仅5KB,轻量易读,适合作为教学补充或项目原型参考。目前已有1034人学习下载,内容紧扣工程实际,提供从射频信号到可读船舶动态信息的完整处理链条,尤其适合需快速掌握AIS数据流闭环、开展接收设备驱动对接或GIS可视化前置开发的开发者与学生。

1. AIS教程、驱动、解码、解析:一条从串口数据到船舶轨迹可视化的完整链路

你手头刚接上一台AIS接收模块,USB线一插,串口有数据涌出——但满屏是!AIVDM,1,1,,B,13uG?w000000000000000000000,0*6C这样的字符串,根本看不出哪条船在哪、航速多少、是否在转向。这不是“没信号”,而是AIS数据流尚未被真正‘唤醒’:它需要正确的硬件驱动加载、符合NMEA 0183协议的串口通信配置、严格遵循ITU-R M.1371标准的二进制帧解码,最后才是把解包后的字段映射成经纬度、COG、SOG等可读信息。本教程不讲抽象协议,只聚焦一线工程师每天真实面对的四道关卡:Linux/Windows下AIS设备识别与驱动加载(尤其国产CH340/CP2102芯片兼容性)、用Python或C实现实时串口收发与帧同步(解决粘包/断帧)、基于bit-level位操作完成VDM/VDO报文的6-bit Base64解码与字段提取(跳过黑匣子库,直击核心逻辑)、最终将原始MMSI+位置+动态字段组装成GeoJSON或CSV供GIS平台消费。适合嵌入式开发者、海事IoT方案集成者、高校海洋感知课程实践者——只要你的目标是“让AIS模块吐出能画航线图的数据”,这篇就是你该停下来的那一页。


2. 驱动层:让操作系统真正“看见”AIS接收器

AIS接收器本质是串口设备,但并非所有USB转串口芯片都能即插即用。常见模块使用CH340、CP2102或FTDI芯片,而驱动缺失会导致ls /dev/tty*无输出、dmesg | grep tty报device descriptor read/64, error -71,或Windows设备管理器中显示“未知设备”。必须分平台确认驱动状态并手动干预。

2.1 Linux下驱动识别与权限修复

多数现代Linux发行版已内置CH340/CP2102驱动,但需验证内核模块是否加载:

# 检查USB设备是否被识别(插入AIS模块后执行) lsusb | grep -i "ch340\|cp210\|ftdi" # 正常应返回类似:Bus 001 Device 005: ID 1a86:7523 QinHeng Electronics HL-340 USB-Serial adapter # 查看内核日志确认驱动绑定 dmesg | tail -20 | grep -i "ch340\|cp210\|tty" # 成功时输出:ch340 1-1.2:1.0: ch341-uart converter detected # 若无此行,说明驱动未加载

若驱动未加载,需手动安装(以Ubuntu/Debian为例):

# 安装CH340驱动(常见于国产AIS模块) sudo apt update && sudo apt install -y build-essential linux-headers-$(uname -r) git clone https://github.com/juliagoda/CH341SER.git cd CH341SER make && sudo make load # 加载后检查/dev/ttyUSB*是否出现 ls /dev/ttyUSB*

提示:权限问题比驱动缺失更常见
即使设备识别成功,普通用户默认无权访问/dev/ttyUSB0。临时方案:sudo chmod a+rw /dev/ttyUSB0;永久方案:将用户加入dialout组
sudo usermod -a -G dialout $USER && newgrp dialout(执行后需重启终端)

2.2 Windows下驱动安装与COM端口锁定

Windows 10/11对CH340支持较好,但部分新版CH340芯片需手动指定.inf文件。若设备管理器中显示黄色感叹号:

  1. 右键“未知设备” → “更新驱动程序” → “浏览我的计算机以查找驱动程序软件”
  2. 下载官方CH340驱动( WCH官网 ),解压后指向CH341SER.INF
  3. 安装完成后,在“端口(COM和LPT)”下找到新分配的COM端口(如COM4)

注意:避免COM端口号漂移
多个USB串口设备接入时,Windows可能每次分配不同COM号。在设备管理器中右键该端口 → “属性” → “端口设置” → “高级” → 手动指定一个高位COM号(如COM10),防止脚本因端口变更而中断。

2.3 验证串口通信:用minicom/screen发送心跳指令

驱动就绪后,需确认AIS模块处于接收模式且串口参数匹配。绝大多数AIS接收器默认波特率38400、8N1(8数据位、无校验、1停止位)、无流控:

# Linux下用screen连接(Ctrl+A, K 退出) screen /dev/ttyUSB0 38400 # 或用minicom(需先配置:minicom -s → Serial port setup → A: /dev/ttyUSB0, E: 38400 8N1) minicom -D /dev/ttyUSB0 -b 38400

连接成功后,应立即看到连续NMEA语句(每秒1~10条):

$GPGGA,082312.00,3114.2912,N,12127.3210,E,1,08,1.1,12.3,M,5.2,M,,*6A !AIVDM,1,1,,B,13uG?w000000000000000000000,0*6C !AIVDM,1,1,,B,40285wp00000000000000000000,0*3C

若无输出,检查:①AIS天线是否接入且放置于开阔窗边;②模块供电是否充足(部分模块需5V@500mA);③串口参数是否误设为9600或115200。


3. 解码层:从NMEA封装到AIS二进制净荷的精准剥离

AIS数据通过NMEA 0183协议传输,但实际有效信息藏在!AIVDM或!AIVDO语句的第6字段(即逗号分隔后的第6项)中。该字段是Base64编码的6-bit二进制数据,需经三步还原:NMEA帧解析 → Base64解码 → 6-bit位流重组 → ITU-R M.1371字段提取。跳过现成库(如pyais)直接手写,才能理解为何13uG?w000000000000000000000解出来是MMSI=219000000、SOG=0.0、COG=51.1°。

3.1 NMEA帧解析:定位VDM/VDO净荷与校验

NMEA语句格式为$[TalkerID][SentenceType],[Field1],[Field2],...,[Checksum]*XX。AIS专用语句为!AIVDM(接收)或!AIVDO(发送),关键字段如下:

字段序号含义示例值说明
1总分片数1单帧报文为1,长报文(如静态信息)可达4片
2当前分片序号1用于重组多片报文
3序列号(空)通常为空,部分设备填充
4通道标识BA信道或B信道,AIS双信道轮询
5Base64净荷13uG?w000000000000000000000核心待解码字段,长度≤69字符
6校验数0分片序号,非CRC校验
7CRC校验*6CNMEA标准校验,*后两位十六进制

Python解析脚本(带校验):

import re def parse_nmea_line(line): """解析单行NMEA,返回净荷字符串及校验结果""" if not line.startswith(('!AIVDM', '!AIVDO')): return None, False # 提取校验值:*后两位 checksum_match = re.search(r'\*([0-9A-Fa-f]{2})$', line) if not checksum_match: return None, False expected_crc = int(checksum_match.group(1), 16) # 计算实际CRC:从!或$开始,到*之前所有字符异或 data_part = line.split('*')[0] actual_crc = 0 for c in data_part[1:]: # 跳过!或$ actual_crc ^= ord(c) if actual_crc != expected_crc: return None, False # 校验失败,丢弃 # 提取第5字段(Base64净荷) fields = line.split(',') if len(fields) < 6: return None, False payload = fields[5].strip() return payload, True # 测试 line = "!AIVDM,1,1,,B,13uG?w000000000000000000000,0*6C" payload, valid = parse_nmea_line(line) print(f"净荷: {payload}, 校验: {valid}") # 净荷: 13uG?w000000000000000000000, 校验: True

逻辑说明:NMEA校验是简单异或,非CRC32。*6C表示从!后第一个字符到*前所有ASCII码异或结果为0x6C。校验失败意味着传输错误或模块故障,必须丢弃,否则解码结果全错。

3.2 Base64解码:6-bit字符集映射与位流重组

AIS Base64非标准RFC 4648,而是ITU-R M.1371定义的64字符表(0-9,A-Z,a-z,@and`)。需构建映射字典:

# AIS Base64字符集(索引0~63) AIS_BASE64_CHARS = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz@`" def ais_base64_decode(payload): """将AIS Base64字符串解码为bytes""" if not payload: return b'' # 将每个字符转为6-bit整数 bits_6 = [] for c in payload: try: idx = AIS_BASE64_CHARS.index(c) bits_6.append(idx) except ValueError: raise ValueError(f"Invalid AIS Base64 char: {c}") # 合并为连续bit流:每个6-bit左移拼接 bit_stream = 0 for i, val in enumerate(bits_6): bit_stream |= (val << (6 * (len(bits_6) - 1 - i))) # 转为bytes:每8-bit一组 byte_len = (len(bits_6) * 6 + 7) // 8 result = bytearray(byte_len) for i in range(byte_len): # 取bit_stream中第i个byte(从高到低) start_bit = (byte_len - 1 - i) * 8 if start_bit < len(bits_6) * 6: # 从bit_stream中提取8位 byte_val = 0 for j in range(8): bit_pos = start_bit + j if bit_pos < len(bits_6) * 6: # 检查bit_stream在bit_pos位是否为1 if bit_stream & (1 << (len(bits_6) * 6 - 1 - bit_pos)): byte_val |= (1 << (7 - j)) result[i] = byte_val return bytes(result) # 测试:解码示例净荷 payload = "13uG?w000000000000000000000" decoded_bytes = ais_base64_decode(payload) print(f"解码bytes: {decoded_bytes.hex()}") # 输出: 01d247bf00000000000000000000000000000000000000000000000000000000

参数说明:payload长度决定解码后bit数(len*6),进而决定bytes长度。示例中28字符*6=168bit→21bytes。注意高位补零规则:AIS规定净荷不足时末尾补0,解码时需按标准长度截取(如类型1报文固定为168bit)。

3.3 位级字段提取:按ITU-R M.1371标准切片

解码得到的bytes是原始二进制位流,需按AIS报文类型(Type 1/2/3为位置报告,Type 5为静态信息)逐bit读取。以Type 1(最常见)为例,结构如下(共168bit):

起始bit长度(bit)字段名说明
06Message ID固定为1
62Repeat indicator重发标志
830MMSI船舶识别码
382Navigation status航行状态(锚泊/航行等)
4012ROT转向率
5210SOG对地航速(0.1节精度)
621Position accuracy定位精度
6328Longitude经度(1/10000分)
9127Latitude纬度(1/10000分)
11812COG航迹向(0.1°精度)
1309HDG船首向(0.1°精度)
1396TimestampUTC时间戳
1452Maneuver indicator操作指示
1473Spare保留位
1501RAIM flagRAIM校验标志
15119Radio status通信状态

Python位提取函数:

def extract_type1_fields(data_bytes): """从168bit Type 1报文bytes中提取关键字段""" if len(data_bytes) < 21: # 168bit = 21bytes raise ValueError("Type 1 data too short") # 将bytes转为bit string(大端,最高位在前) bit_str = ''.join(f'{b:08b}' for b in data_bytes) if len(bit_str) > 168: bit_str = bit_str[:168] # 截断多余位 elif len(bit_str) < 168: bit_str = bit_str.zfill(168) # 左侧补零 def get_bits(start, length): return bit_str[start:start+length] # 按表提取 mmsi = int(get_bits(8, 30), 2) sog = int(get_bits(52, 10), 2) * 0.1 # 单位:节 lon_raw = int(get_bits(63, 28), 2) lat_raw = int(get_bits(91, 27), 2) # 经纬度转换:ITU-R M.1371规定,lon范围-180~180,lat范围-90~90 # lon: 28bit有符号,需处理补码 if lon_raw & 0x08000000: # 最高位为1,负数 lon_deg = (lon_raw - 0x10000000) / 600000.0 else: lon_deg = lon_raw / 600000.0 if lat_raw & 0x04000000: # 27bit,最高位为1 lat_deg = (lat_raw - 0x08000000) / 600000.0 else: lat_deg = lat_raw / 600000.0 cog = int(get_bits(118, 12), 2) * 0.1 hdg = int(get_bits(130, 9), 2) * 0.1 if int(get_bits(130, 9), 2) < 511 else None return { 'mmsi': mmsi, 'sog': sog, 'lon': round(lon_deg, 6), 'lat': round(lat_deg, 6), 'cog': cog, 'hdg': hdg } # 测试 fields = extract_type1_fields(decoded_bytes) print(fields) # {'mmsi': 219000000, 'sog': 0.0, 'lon': 121.45535, 'lat': 31.238187, 'cog': 51.1, 'hdg': 51.1}

血泪经验:经纬度转换极易翻车!必须用补码逻辑,不能直接除。lon_raw=0x0FFFFFFF看似很大,实为-0.000001°,直接/600000会得正数。务必检查最高位再决定加减偏移量。


4. 解析层:从原始字段到时空轨迹的工程化组织

解出单条报文的MMSI、经纬度、SOG、COG后,真正的挑战才开始:如何区分同一MMSI的多条报文?如何剔除GPS跳变噪声?如何关联静态信息(船名、呼号)?如何输出为GIS友好的格式?这一步决定数据能否进入生产系统。

4.1 MMSI会话管理:用滑动窗口聚合轨迹点

单条AIS报文无时间戳(仅UTC分钟内秒数),需结合系统时间打标。但更关键的是按MMSI聚合同一会话,避免不同船舶数据混杂:

from collections import defaultdict, deque import time class AISTrajectoryManager: def __init__(self, max_gap_seconds=30, max_points=1000): # {mmsi: deque of (timestamp, fields)} self.sessions = defaultdict(lambda: deque(maxlen=max_points)) self.max_gap = max_gap_seconds def add_point(self, mmsi, fields): now = time.time() # 若该MMSI无历史记录,或最新点时间差过大,则新建会话 if not self.sessions[mmsi] or (now - self.sessions[mmsi][-1][0] > self.max_gap): self.sessions[mmsi].clear() self.sessions[mmsi].append((now, fields)) def get_latest_trajectory(self, mmsi, n=10): """获取某MMSI最近n个点的轨迹""" if mmsi not in self.sessions or len(self.sessions[mmsi]) == 0: return [] return [point[1] for point in list(self.sessions[mmsi])[-n:]] # 使用示例 mgr = AISTrajectoryManager() mgr.add_point(219000000, {'lon': 121.455, 'lat': 31.238, 'sog': 0.0}) mgr.add_point(219000000, {'lon': 121.456, 'lat': 31.239, 'sog': 0.1}) print(mgr.get_latest_trajectory(219000000))

为什么需要会话管理?
AIS模块可能同时收到多艘船数据,若不按MMSI隔离,轨迹线会交叉混乱。max_gap_seconds=30是经验值:船舶静止时AIS每3~10秒发一次,若超30秒无新点,视为会话中断,防止旧轨迹污染新航线。

4.2 噪声过滤:基于速度与距离的卡尔曼预滤波

AIS GPS存在跳变(如城市峡谷多径效应),直接绘图会出现“瞬移”伪影。简单但有效的过滤策略:

过滤类型触发条件处理方式
速度突变`sog_current - sog_prev
位置突变haversine_distance(prev, current) > 1km丢弃当前点
静止误报sog < 0.2节且 `lon_diff
import math def haversine_distance(p1, p2): """计算两点间球面距离(米)""" R = 6371000 # 地球半径(米) lat1, lon1 = math.radians(p1['lat']), math.radians(p1['lon']) lat2, lon2 = math.radians(p2['lat']), math.radians(p2['lon']) dlat = lat2 - lat1 dlon = lon2 - lon1 a = math.sin(dlat/2)**2 + math.cos(lat1)*math.cos(lat2)*math.sin(dlon/2)**2 c = 2 * math.asin(math.sqrt(a)) return R * c def filter_ais_point(prev_point, curr_point, max_speed_diff=5.0, max_dist_m=1000.0): """过滤单点噪声""" if not prev_point: return curr_point # 速度突变 if abs(curr_point['sog'] - prev_point['sog']) > max_speed_diff: return prev_point # 位置突变 dist = haversine_distance(prev_point, curr_point) if dist > max_dist_m: return prev_point return curr_point # 使用 prev = {'lat': 31.238, 'lon': 121.455, 'sog': 0.0} curr = {'lat': 31.245, 'lon': 121.460, 'sog': 0.1} # 距离约780m,合理 clean = filter_ais_point(prev, curr)

玄学参数:max_dist_m=1000是平衡灵敏度与鲁棒性的关键。设太小(如100m)会过度平滑真实机动;设太大(如5000m)则无法拦截GPS跳变。建议用实测数据调整:抓取1小时AIS流,统计正常船舶最大瞬时位移,取P95值。

4.3 静态信息关联:Type 5报文与MMSI绑定

Type 5报文含船名、呼号、船型等静态信息,但发送频率远低于Type 1(约每6分钟1次)。需建立MMSI→静态信息缓存:

class AISStaticDB: def __init__(self): self.cache = {} # {mmsi: {name, callsign, shiptype, ...}} self.last_update = {} # {mmsi: timestamp} def update_from_type5(self, mmsi, fields): """从Type 5报文更新静态信息""" self.cache[mmsi] = { 'name': fields.get('name', '').strip('\x00'), 'callsign': fields.get('callsign', '').strip('\x00'), 'shiptype': fields.get('shiptype', 0), 'length': fields.get('length', 0), 'width': fields.get('width', 0) } self.last_update[mmsi] = time.time() def get_static(self, mmsi): """获取静态信息,超时则返回None""" if mmsi not in self.last_update: return None if time.time() - self.last_update[mmsi] > 3600: # 1小时未更新,过期 return None return self.cache.get(mmsi, None) # 在主循环中调用 static_db = AISStaticDB() # 解析到Type 5时: # static_db.update_from_type5(mmsi, type5_fields) # 解析Type 1时: static_info = static_db.get_static(219000000) if static_info: print(f"船名: {static_info['name']}, 呼号: {static_info['callsign']}")

避坑:Type 5报文分片重组
Type 5长达424bit,需2片VDM传输(字段1=2)。必须缓存第一片,待第二片到达后拼接payload1+payload2再解码,否则解出乱码。


5. 避坑指南:AIS数据链路上的5个致命陷阱

AIS项目落地中最耗时的往往不是编码,而是排查那些“看起来正常却输出错误”的隐性故障。以下是某跨平台AIS系统开发中踩过的5个典型坑,按现象→原因→解决结构整理,每一条都来自真实翻车现场。

5.1 现象:串口持续输出!AIVDM,0,0,,B,00000000000,0*40这类全零报文

原因:AIS模块天线未接入或短路,导致内部AGC(自动增益控制)饱和,输出空帧。部分模块在无信号时会周期性发送占位符。
解决:

  • 用万用表测天线接口阻抗(应为50Ω),排除短路;
  • 将天线移至窗外开阔处,观察!AIVDM出现频率是否从1Hz升至5~10Hz;
  • 若仍无效,尝试更换天线(AIS专用天线中心频点161.975MHz/162.025MHz,非GPS天线)。

5.2 现象:解码出的经纬度全部为0.000000或极大值(如180.000000)

原因:位提取时未处理补码。ITU-R M.1371规定经纬度为有符号28/27bit,但直接int(...,2)会当无符号处理。
解决:

  • 经度:检查第28bit(bit_str[62]),若为1,则lon_raw -= 0x10000000;
  • 纬度:检查第27bit(bit_str[90]),若为1,则lat_raw -= 0x08000000;
  • 再除以600000.0。务必在extract_type1_fields中加入此逻辑,不可省略。

5.3 现象:同一MMSI的轨迹点在地图上呈放射状散开(“星爆”效应)

原因:未做时间同步,系统时间与AIS模块UTC时间偏差大,导致多源数据时间轴错位。AIS报文内只有Timestamp(0~59秒),无日期,依赖主机时间打标。
解决:

  • 启动时用NTP校准系统时间:sudo timedatectl set-ntp true;
  • 在add_point前,用time.time()而非time.clock();
  • 对高精度需求场景,改用PPS信号同步(需硬件支持)。

5.4 现象:Python脚本运行数小时后内存暴涨,最终OOM崩溃

原因:AISTrajectoryManager.sessions无限增长,虽设maxlen,但defaultdict的key不会自动删除,长期运行后sessions字典本身膨胀。
解决:

  • 增加定期清理:每1000次add_point后,遍历sessions,删除len(deque)==0的key;
  • 或改用weakref.WeakValueDictionary,但需确保session对象不被其他引用持有。

5.5 现象:Windows下pyserial读取串口时CPU占用100%,readline()阻塞

原因:timeout=0(非阻塞)或timeout=None(永久阻塞)配置不当,导致循环空转或死锁。AIS数据是流式到达,非固定包长。
解决:

  • 串口初始化必须设timeout=0.1(100ms);
  • 读取时用read(1024)而非readline(),因AIS无可靠换行符;
  • 在循环中加time.sleep(0.01)防CPU飙高。

6. 进阶技巧:用GeoJSON实时推送与Web可视化闭环

当AIS数据能稳定解码并过滤后,终极价值在于“让数据动起来”。我一般会用最小成本搭建一个浏览器可直连的实时轨迹看板:Python后端持续解析AIS流,生成GeoJSON FeatureCollection,前端用Leaflet每2秒拉取并刷新标记。不依赖任何商业GIS平台,纯开源栈。

6.1 后端:Flask提供GeoJSON API端点

from flask import Flask, jsonify import json app = Flask(__name__) # 全局共享轨迹管理器(生产环境应换Redis) trajectory_mgr = AISTrajectoryManager(max_points=500) static_db = AISStaticDB() @app.route('/api/ais/geojson') def get_geojson(): features = [] for mmsi, points in trajectory_mgr.sessions.items(): if len(points) < 2: continue # 取最新点作为标记位置 latest = points[-1][1] static = static_db.get_static(mmsi) or {} feature = { "type": "Feature", "properties": { "mmsi": mmsi, "name": static.get('name', f"MMSI-{mmsi}"), "sog": latest['sog'], "cog": latest['cog'], "timestamp": points[-1][0] }, "geometry": { "type": "Point", "coordinates": [latest['lon'], latest['lat']] } } features.append(feature) return jsonify({ "type": "FeatureCollection", "features": features }) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=False) # 生产环境禁用debug

部署提示:用gunicorn替代flask run:gunicorn --bind 0.0.0.0:5000 --workers 2 app:app,避免单线程阻塞。

6.2 前端:Leaflet每2秒拉取并更新标记

<!DOCTYPE html> <html> <head> <link rel="stylesheet" href="https://unpkg.com/leaflet@1.9.4/dist/leaflet.css"/> </head> <body> <div id="map" style="height: 600px;"></div> <script src="https://unpkg.com/leaflet@1.9.4/dist/leaflet.js"></script> <script> const map = L.map('map').setView([31.2, 121.4], 10); L.tileLayer('https://{a-d}.tile.openstreetmap.org/{z}/{x}/{y}.png').addTo(map); let markers = {}; function updateMap() { fetch('/api/ais/geojson') .then(r => r.json()) .then(data => { // 清除旧标记 Object.values(markers).forEach(m => map.removeLayer(m)); markers = {}; // 添加新标记 data.features.forEach(f => { const pos = f.geometry.coordinates; const m = L.marker([pos[1], pos[0]]).addTo(map); m.bindPopup(` <b>${f.properties.name}</b><br> MMSI: ${f.properties.mmsi}<br> SOG: ${f.properties.sog} kn<br> COG: ${f.properties.cog}° `); markers[f.properties.mmsi] = m; <p> <a href="https://download.csdn.net/download/luopeiyuan1990/4652100" style="color:#ec7500;font-size:14px;"> 本文还有配套的精品资源,点击获取 </a> <img alt="menu-r.4af5f7ec.gif" src="https://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif" style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;"> </p>
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进