news 2026/5/13 4:26:32

别再对着乱码发愁了！手把手教你用Python解码AIS VDM暗码（附完整代码）

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
别再对着乱码发愁了！手把手教你用Python解码AIS VDM暗码（附完整代码）

从AIS暗码到可读数据：Python实战解析指南

当你第一次看到类似!AIVDM,1,1,,A,169DvlgP1R8KPtvFBfOCt3?h0@RT,0*03这样的字符串时，可能会感到一头雾水。这串看似随机的字符实际上是AIS(船舶自动识别系统)传输的VDM(VHF Data-link Message)报文，包含了船舶位置、航速、航向等关键信息。本文将带你用Python一步步解开这些"暗码"，将其转化为结构化的可读数据。

1. AIS VDM报文基础解析

AIS系统通过VHF无线电广播船舶信息，采用NMEA 0183标准格式传输。一条完整的AIS VDM报文通常由以下部分组成：

!AIVDM,1,1,,A,169DvlgP1R8KPtvFBfOCt3?h0@RT,0*03

让我们分解这个报文的各个部分：

  • !AIVDM：报文类型标识
  • 1：当前片段总数
  • 1：当前片段序号
  • ``：连续报文标识(空表示单条完整报文)
  • A：信道标识(A或B)
  • 169DvlgP1R8KPtvFBfOCt3?h0@RT：实际载荷数据
  • 0：填充位数
  • 03：校验和

注意：多片段报文在海上通信中很常见，需要正确重组才能解析完整信息。

2. 搭建Python解析环境

在开始解码前，我们需要准备合适的Python环境。推荐使用Python 3.8+版本，并安装以下关键库：

pip install pyais pynmea2 bitstring

这些库将帮助我们：

  • pyais：专业的AIS解码库
  • pynmea2：处理NMEA 0183格式
  • bitstring：处理二进制数据

创建一个新的Python文件，导入必要的库：

import pyais from pynmea2 import parse from bitstring import BitArray import math

3. 解析NMEA报文结构

首先，我们需要从原始字符串中提取出有效载荷部分。使用pynmea2库可以轻松完成这一任务：

def parse_nmea(nmea_str): try: msg = parse(nmea_str) return { 'message_type': msg.sentence_type, 'fragment_count': msg.num_fragments, 'fragment_number': msg.fragment_num, 'sequential_id': msg.sequential_id, 'channel': msg.channel, 'payload': msg.data, 'padding': msg.padding, 'checksum': msg.checksum } except Exception as e: print(f"解析NMEA失败: {e}") return None

测试这个函数：

nmea_str = "!AIVDM,1,1,,A,169DvlgP1R8KPtvFBfOCt3?h0@RT,0*03" parsed = parse_nmea(nmea_str) print(parsed)

输出将是一个包含各个字段的字典，这样我们就完成了第一步的结构化处理。

4. 6-bit ASCII解码技术

AIS载荷使用6-bit ASCII编码，这是一种特殊的编码方式，将8-bit字符压缩为6-bit，以提高传输效率。我们需要编写解码函数：

def decode_6bit_ascii(payload): # 6-bit ASCII字符集 chars = "@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^- !\"#$%&'()*+,-./0123456789:;<=>?" # 移除填充位(如果有) if 'padding' in payload and payload['padding'] > 0: payload_str = payload['payload'][:-1] # 移除最后一个字符 else: payload_str = payload['payload'] # 将每个字符转换为6-bit二进制 bit_string = "" for char in payload_str: if char in chars: value = chars.index(char) bit_string += f"{value:06b}" # 6位二进制表示 else: raise ValueError(f"无效的6-bit ASCII字符: {char}") return bit_string

这个函数将返回一个二进制字符串，例如对于我们的示例报文，会得到类似0101101001...这样的输出。

5. 提取AIS消息字段

获得二进制数据后，我们需要根据AIS消息类型解析具体字段。AIS有27种消息类型，最常见的是位置报告(类型1-3)。下面是一个解析位置报告的示例：

def parse_position_report(bit_string): bits = BitArray(bin=bit_string) # 读取固定位置字段 message_type = bits[0:6].uint repeat_indicator = bits[6:8].uint mmsi = bits[8:38].uint navigation_status = bits[38:42].uint rate_of_turn = bits[42:50].int speed_over_ground = bits[50:60].uint * 0.1 # 转换为节 position_accuracy = bits[60:61].uint longitude = bits[61:89].int * 1.0 / 600000 # 转换为度 latitude = bits[89:116].int * 1.0 / 600000 # 转换为度 course_over_ground = bits[116:128].uint * 0.1 # 转换为度 true_heading = bits[128:137].uint timestamp = bits[137:143].uint maneuver_indicator = bits[143:145].uint spare = bits[145:148].uint raim_flag = bits[148:149].uint radio_status = bits[149:168].uint return { 'message_type': message_type, 'repeat_indicator': repeat_indicator, 'mmsi': mmsi, 'navigation_status': navigation_status, 'rate_of_turn': rate_of_turn, 'speed_over_ground': speed_over_ground, 'position_accuracy': position_accuracy, 'longitude': longitude, 'latitude': latitude, 'course_over_ground': course_over_ground, 'true_heading': true_heading, 'timestamp': timestamp, 'maneuver_indicator': maneuver_indicator, 'raim_flag': raim_flag, 'radio_status': radio_status }

6. 处理特殊值和边界情况

AIS数据中有一些特殊值需要特别注意：

  • 经度/纬度：值为181度(经度)或91度(纬度)表示数据不可用
  • 航速：102.3节表示数据不可用或大于102.2节
  • 航向：511度表示数据不可用
  • 船首向：511度表示数据不可用

我们需要在解析函数中添加对这些特殊值的处理：

def adjust_special_values(data): if data['longitude'] >= 181: data['longitude'] = None if data['latitude'] >= 91: data['latitude'] = None if data['speed_over_ground'] >= 102.3: data['speed_over_ground'] = None if data['course_over_ground'] >= 360: data['course_over_ground'] = None if data['true_heading'] == 511: data['true_heading'] = None return data

7. 完整解析流程示例

现在，我们将所有步骤组合起来，形成一个完整的解析流程：

def decode_ais_message(nmea_str): # 1. 解析NMEA结构 nmea_data = parse_nmea(nmea_str) if not nmea_data: return None # 2. 解码6-bit ASCII try: bit_string = decode_6bit_ascii(nmea_data) except ValueError as e: print(f"解码6-bit ASCII失败: {e}") return None # 3. 解析消息类型 message_type = BitArray(bin=bit_string[:6]).uint # 4. 根据类型调用不同解析器 if message_type in (1, 2, 3): # 位置报告 data = parse_position_report(bit_string) elif message_type == 5: # 静态和航程相关数据 data = parse_static_voyage_data(bit_string) else: print(f"不支持的消息类型: {message_type}") return None # 5. 处理特殊值 data = adjust_special_values(data) return data

8. 使用pyais库简化流程

虽然手动解析有助于理解原理，但在实际项目中，我们可以使用专业的pyais库来简化流程：

from pyais import decode def decode_with_pyais(nmea_str): try: decoded = decode(nmea_str) return decoded.asdict() except Exception as e: print(f"解码失败: {e}") return None

这个函数可以直接返回结构化的AIS数据，包含了所有字段和适当的类型转换。

9. 处理多片段报文

在实际应用中，AIS报文可能被分割成多个片段传输。我们需要先重组这些片段，然后再解析：

from collections import defaultdict class AISReassembler: def __init__(self): self.fragments = defaultdict(dict) def add_fragment(self, nmea_str): msg = parse_nmea(nmea_str) if not msg: return None key = (msg['sequential_id'], msg['channel']) self.fragments[key][msg['fragment_number']] = msg # 检查是否收集了所有片段 if len(self.fragments[key]) == msg['fragment_count']: # 按顺序拼接片段 sorted_frags = [self.fragments[key][i] for i in range(1, msg['fragment_count']+1)] combined_payload = ''.join(f['payload'] for f in sorted_frags) # 创建组合后的NMEA字符串(使用第一个片段的元数据) first = sorted_frags[0] reassembled = f"!{first['message_type']},{first['fragment_count']},1,{first['sequential_id']},{first['channel']},{combined_payload},{first['padding']}*{first['checksum']}" # 移除已完成的片段 del self.fragments[key] return reassembled return None

使用示例：

reassembler = AISReassembler() # 假设我们有两个片段 fragment1 = "!AIVDM,2,1,1,A,55?MbV02;H;s<HtKR20EHE:0@T4@Dn2222222216L961O5Gf0NSQEp6ClRp8,0*1C" fragment2 = "!AIVDM,2,2,1,A,88888888880,2*25" # 添加片段 reassembler.add_fragment(fragment1) complete = reassembler.add_fragment(fragment2) if complete: decoded = decode_ais_message(complete) print(decoded)

10. 性能优化与批量处理

当需要处理大量AIS数据时，性能变得很重要。以下是一些优化建议：

  1. 使用生成器处理流数据：
def process_ais_stream(stream): for line in stream: line = line.strip() if line.startswith('!AIVDM'): try: yield decode_ais_message(line) except Exception as e: print(f"处理失败: {line} - {e}") continue
  1. 多线程处理：
from concurrent.futures import ThreadPoolExecutor def bulk_decode(nmea_strings, max_workers=4): with ThreadPoolExecutor(max_workers=max_workers) as executor: results = list(executor.map(decode_ais_message, nmea_strings)) return [r for r in results if r is not None]
  1. 缓存解码结果：
from functools import lru_cache @lru_cache(maxsize=1000) def cached_decode(nmea_str): return decode_ais_message(nmea_str)

11. 数据验证与错误处理

健壮的解析器需要包含完善的错误处理机制：

def safe_decode(nmea_str): if not nmea_str.startswith('!AIVDM'): raise ValueError("不是AIVDM报文") try: # 校验和验证 parts = nmea_str.split('*') if len(parts) != 2: raise ValueError("无效的NMEA格式") calculated = 0 for c in parts[0][1:]: # 跳过起始的'!' calculated ^= ord(c) checksum = int(parts[1], 16) if calculated != checksum: raise ValueError(f"校验和不匹配: 计算值{calculated:02X}, 报文值{checksum:02X}") return decode_ais_message(nmea_str) except Exception as e: print(f"安全解码失败: {e}") return None

12. 将解析结果转换为GeoJSON

为了在地图上可视化AIS数据，我们可以将其转换为GeoJSON格式：

import json def to_geojson(ais_data): if not ais_data.get('latitude') or not ais_data.get('longitude'): return None feature = { "type": "Feature", "geometry": { "type": "Point", "coordinates": [ais_data['longitude'], ais_data['latitude']] }, "properties": { "mmsi": ais_data['mmsi'], "sog": ais_data['speed_over_ground'], "cog": ais_data['course_over_ground'], "heading": ais_data['true_heading'], "status": ais_data['navigation_status'], "timestamp": ais_data['timestamp'] } } return feature

13. 实战案例：实时AIS数据监控系统

结合以上技术，我们可以构建一个简单的实时AIS监控系统：

import socket from datetime import datetime class AISMonitor: def __init__(self, host, port): self.host = host self.port = port self.reassembler = AISReassembler() def start(self): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((self.host, self.port)) while True: data = s.recv(4096).decode('ascii') if not data: break for line in data.splitlines(): line = line.strip() if line.startswith('!AIVDM'): complete = self.reassembler.add_fragment(line) if complete: msg = decode_ais_message(complete) if msg: self.process_message(msg) def process_message(self, msg): timestamp = datetime.now().isoformat() print(f"[{timestamp}] MMSI: {msg['mmsi']}, " f"Position: ({msg['latitude']:.4f}, {msg['longitude']:.4f}), " f"SOG: {msg['speed_over_ground']} knots, " f"COG: {msg['course_over_ground']}°")

使用示例：

monitor = AISMonitor('ais.server.example.com', 1234) monitor.start()

14. 常见问题与调试技巧

在解析AIS数据时，可能会遇到各种问题。以下是一些常见问题及其解决方法：

  1. 校验和错误：

    • 检查原始数据是否被截断或损坏
    • 验证校验和计算是否正确

  2. 解码后字段值不合理：

    • 确认是否正确处理了6-bit ASCII到二进制的转换
    • 检查字段的起始位置和长度是否正确
    • 验证特殊值(如511度)是否被正确处理

  3. 多片段报文无法重组：

    • 确保所有片段具有相同的sequential_id和channel
    • 检查片段编号是否连续
    • 确认是否收到了所有片段

调试时可以添加详细的日志记录：

import logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger('ais_parser') def debug_decode(nmea_str): logger.debug(f"原始NMEA: {nmea_str}") nmea_data = parse_nmea(nmea_str) logger.debug(f"NMEA结构: {nmea_data}") bit_string = decode_6bit_ascii(nmea_data) logger.debug(f"二进制字符串: {bit_string}") message_type = BitArray(bin=bit_string[:6]).uint logger.debug(f"消息类型: {message_type}") # 其余解析步骤...

15. 进阶主题：AIS消息类型扩展

除了基本的位置报告，AIS还定义了多种消息类型。我们可以扩展解析器以支持更多类型：

def parse_static_voyage_data(bit_string): bits = BitArray(bin=bit_string) # 类型5特有字段 ais_version = bits[38:40].uint imo_number = bits[40:70].uint call_sign = bits[70:112].bin # 需要特殊解码 vessel_name = bits[112:232].bin # 需要特殊解码 ship_type = bits[232:240].uint dimension_a = bits[240:249].uint dimension_b = bits[249:258].uint dimension_c = bits[258:264].uint dimension_d = bits[264:270].uint fix_type = bits[270:274].uint eta_month = bits[274:278].uint eta_day = bits[278:283].uint eta_hour = bits[283:288].uint eta_minute = bits[288:294].uint draught = bits[294:302].uint * 0.1 destination = bits[302:422].bin # 需要特殊解码 dte = bits[422:423].uint spare = bits[423:424].uint # 解码文本字段 def decode_ais_text(bin_str): chars = "@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^- !\"#$%&'()*+,-./0123456789:;<=>?" text = "" for i in range(0, len(bin_str), 6): chunk = bin_str[i:i+6] if len(chunk) < 6: break index = int(chunk, 2) if index < len(chars): text += chars[index] return text.strip() return { 'message_type': 5, 'ais_version': ais_version, 'imo_number': imo_number, 'call_sign': decode_ais_text(call_sign), 'vessel_name': decode_ais_text(vessel_name), 'ship_type': ship_type, 'dimensions': { 'A': dimension_a, 'B': dimension_b, 'C': dimension_c, 'D': dimension_d }, 'fix_type': fix_type, 'eta': f"{eta_month}-{eta_day} {eta_hour}:{eta_minute}", 'draught': draught, 'destination': decode_ais_text(destination), 'dte': dte }
版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/5/13 4:21:21

K8sGPT：AI驱动的Kubernetes智能诊断与根因分析实践指南

1. 项目概述&#xff1a;当Kubernetes运维遇上AI副驾驶如果你是一名Kubernetes&#xff08;K8s&#xff09;的运维工程师、SRE或者开发者&#xff0c;那么下面这个场景你一定不陌生&#xff1a;凌晨三点&#xff0c;告警响了&#xff0c;你睡眼惺忪地爬起来&#xff0c;登录集群…

作者头像 李华
网站建设 2026/5/13 4:18:32

Ice：终极macOS菜单栏管理器 - 释放你的菜单栏空间

Ice&#xff1a;终极macOS菜单栏管理器 - 释放你的菜单栏空间 【免费下载链接】Ice Powerful menu bar manager for macOS 项目地址: https://gitcode.com/GitHub_Trending/ice/Ice Ice是一款功能强大的macOS菜单栏管理工具&#xff0c;它不仅能帮助你隐藏和显示菜单栏项…

作者头像 李华
网站建设 2026/5/13 4:12:09

Convox Rack故障排除指南：常见问题与解决方案大全

Convox Rack故障排除指南&#xff1a;常见问题与解决方案大全 【免费下载链接】rack Private PaaS built on native AWS services for maximum privacy and minimum upkeep 项目地址: https://gitcode.com/gh_mirrors/rack/rack Convox Rack作为基于AWS原生服务构建的私…

作者头像 李华
网站建设 2026/5/13 4:09:09

Node.js后端开发入门：learning-article项目中的全栈技术栈完整指南

Node.js后端开发入门&#xff1a;learning-article项目中的全栈技术栈完整指南 【免费下载链接】learning-article 学习资源 or 大前端导航&#xff0c;持续更新 项目地址: https://gitcode.com/gh_mirrors/le/learning-article 想要快速掌握Node.js后端开发并构建完整的…

作者头像 李华