299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""
|
|
AIS vessel decoder — Types 1/2/3, 5, 18, 19, 24A/24B.
|
|
Returns dicts compatible with process_message() in main.py.
|
|
"""
|
|
from __future__ import annotations
|
|
from datetime import datetime
|
|
|
|
SHIP_TYPE_NAMES: dict[int, str] = {
|
|
0: "Not available", 20: "WIG", 30: "Fishing", 31: "Towing",
|
|
32: "Towing >200m", 33: "Dredging", 34: "Diving ops",
|
|
35: "Military", 36: "Sailing", 37: "Pleasure craft",
|
|
40: "High speed craft", 50: "Pilot vessel",
|
|
51: "Search and rescue", 52: "Tug", 53: "Port tender",
|
|
54: "Anti-pollution", 55: "Law enforcement",
|
|
58: "Medical transport", 60: "Passenger",
|
|
61: "Passenger haz-A", 62: "Passenger haz-B",
|
|
69: "Passenger (other)", 70: "Cargo",
|
|
71: "Cargo haz-A", 72: "Cargo haz-B",
|
|
79: "Cargo (other)", 80: "Tanker",
|
|
81: "Tanker haz-A", 82: "Tanker haz-B",
|
|
89: "Tanker (other)", 90: "Other", 99: "Other (no info)",
|
|
}
|
|
|
|
NAV_STATUS_NAMES: dict[int, str] = {
|
|
0: "Under way using engine", 1: "At anchor",
|
|
2: "Not under command", 3: "Restricted manoeuvrability",
|
|
4: "Constrained by draught", 5: "Moored", 6: "Aground",
|
|
7: "Engaged in fishing", 8: "Under way sailing",
|
|
9: "Reserved (HSC)", 10: "Reserved (WIG)",
|
|
11: "Power-driven towing astern", 12: "Power-driven pushing ahead",
|
|
13: "Reserved", 14: "AIS-SART/MOB/EPIRB", 15: "Undefined",
|
|
}
|
|
|
|
|
|
def _bits(payload: str, start: int, length: int) -> int:
|
|
acc = 0
|
|
for i in range(start, start + length):
|
|
if i >= len(payload) * 6:
|
|
return 0
|
|
char_idx = i // 6
|
|
bit_idx = 5 - (i % 6)
|
|
char_val = ord(payload[char_idx]) - 48
|
|
if char_val > 39:
|
|
char_val -= 8
|
|
acc = (acc << 1) | ((char_val >> bit_idx) & 1)
|
|
return acc
|
|
|
|
|
|
def _signed(val: int, bits: int) -> int:
|
|
if val >= (1 << (bits - 1)):
|
|
val -= (1 << bits)
|
|
return val
|
|
|
|
|
|
def _text6(payload: str, start: int, length: int) -> str:
|
|
chars = []
|
|
for i in range(length // 6):
|
|
v = _bits(payload, start + i * 6, 6)
|
|
if v < 32:
|
|
v += 64
|
|
if v == 64:
|
|
break
|
|
chars.append(chr(v))
|
|
return ''.join(chars).strip('@').strip()
|
|
|
|
|
|
def decode_type123(payload: str) -> dict | None:
|
|
"""Type 1, 2, 3 — Class A CNB position report."""
|
|
try:
|
|
msg_type = _bits(payload, 0, 6)
|
|
if msg_type not in (1, 2, 3):
|
|
return None
|
|
nav_status = _bits(payload, 6, 4)
|
|
rot_raw = _signed(_bits(payload, 10, 8), 8)
|
|
sog_raw = _bits(payload, 50, 10)
|
|
lon_raw = _signed(_bits(payload, 61, 28), 28)
|
|
lat_raw = _signed(_bits(payload, 89, 27), 27)
|
|
cog_raw = _bits(payload, 116, 12)
|
|
heading = _bits(payload, 128, 9)
|
|
mmsi = _bits(payload, 8, 30)
|
|
|
|
lon = lon_raw / 600000.0
|
|
lat = lat_raw / 600000.0
|
|
sog = sog_raw / 10.0
|
|
cog = cog_raw / 10.0
|
|
|
|
if lat == 0.0 and lon == 0.0:
|
|
return None
|
|
if abs(lat) > 90 or abs(lon) > 180:
|
|
return None
|
|
|
|
rot = None
|
|
if rot_raw not in (0, 127, -128):
|
|
rot = round((rot_raw / 4.733) ** 2 * (1 if rot_raw > 0 else -1), 1)
|
|
|
|
return {
|
|
"type": "vessel",
|
|
"msg_type": msg_type,
|
|
"mmsi": str(mmsi),
|
|
"nav_status": nav_status,
|
|
"nav_status_name": NAV_STATUS_NAMES.get(nav_status, "Unknown"),
|
|
"rot": rot,
|
|
"sog": sog,
|
|
"lat": round(lat, 6),
|
|
"lon": round(lon, 6),
|
|
"cog": cog,
|
|
"heading": heading if heading != 511 else None,
|
|
"fix_type": _bits(payload, 148, 4),
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"source": "AIS",
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def decode_type5(payload: str) -> dict | None:
|
|
"""Type 5 — Class A static + voyage data."""
|
|
try:
|
|
msg_type = _bits(payload, 0, 6)
|
|
if msg_type != 5:
|
|
return None
|
|
mmsi = _bits(payload, 8, 30)
|
|
imo = _bits(payload, 40, 30)
|
|
callsign = _text6(payload, 70, 42)
|
|
name = _text6(payload, 112, 120)
|
|
ship_type = _bits(payload, 232, 8)
|
|
to_bow = _bits(payload, 240, 9)
|
|
to_stern = _bits(payload, 249, 9)
|
|
to_port = _bits(payload, 258, 6)
|
|
to_stbd = _bits(payload, 264, 6)
|
|
# ETA: month(4) day(5) hour(5) minute(6)
|
|
eta_month = _bits(payload, 274, 4)
|
|
eta_day = _bits(payload, 278, 5)
|
|
eta_hour = _bits(payload, 283, 5)
|
|
eta_minute = _bits(payload, 288, 6)
|
|
draught = _bits(payload, 294, 8) / 10.0
|
|
dest = _text6(payload, 302, 120)
|
|
|
|
length = to_bow + to_stern
|
|
beam = to_port + to_stbd
|
|
|
|
eta_str = None
|
|
if eta_month and eta_day:
|
|
eta_str = f"{eta_month:02d}-{eta_day:02d} {eta_hour:02d}:{eta_minute:02d}"
|
|
|
|
return {
|
|
"type": "vessel_static",
|
|
"mmsi": str(mmsi),
|
|
"imo": str(imo) if imo else None,
|
|
"callsign": callsign or None,
|
|
"nombre": name or None,
|
|
"tipo": ship_type,
|
|
"tipo_nombre": SHIP_TYPE_NAMES.get(ship_type, f"Type {ship_type}"),
|
|
"length": length if length else None,
|
|
"beam": beam if beam else None,
|
|
"to_bow": to_bow if to_bow else None,
|
|
"to_stern": to_stern if to_stern else None,
|
|
"to_port": to_port if to_port else None,
|
|
"to_starboard": to_stbd if to_stbd else None,
|
|
"draught": draught if draught else None,
|
|
"eta": eta_str,
|
|
"destino": dest or None,
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"source": "AIS",
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def decode_type18(payload: str) -> dict | None:
|
|
"""Type 18 — Class B standard position report."""
|
|
try:
|
|
msg_type = _bits(payload, 0, 6)
|
|
if msg_type != 18:
|
|
return None
|
|
mmsi = _bits(payload, 8, 30)
|
|
sog_raw = _bits(payload, 46, 10)
|
|
lon_raw = _signed(_bits(payload, 57, 28), 28)
|
|
lat_raw = _signed(_bits(payload, 85, 27), 27)
|
|
cog_raw = _bits(payload, 112, 12)
|
|
heading = _bits(payload, 124, 9)
|
|
|
|
lon = lon_raw / 600000.0
|
|
lat = lat_raw / 600000.0
|
|
sog = sog_raw / 10.0
|
|
cog = cog_raw / 10.0
|
|
|
|
if lat == 0.0 and lon == 0.0:
|
|
return None
|
|
if abs(lat) > 90 or abs(lon) > 180:
|
|
return None
|
|
|
|
return {
|
|
"type": "vessel",
|
|
"msg_type": 18,
|
|
"mmsi": str(mmsi),
|
|
"sog": sog,
|
|
"lat": round(lat, 6),
|
|
"lon": round(lon, 6),
|
|
"cog": cog,
|
|
"heading": heading if heading != 511 else None,
|
|
"nav_status": 0,
|
|
"nav_status_name": "Under way using engine",
|
|
"rot": None,
|
|
"fix_type": _bits(payload, 134, 4),
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"source": "AIS",
|
|
"class_b": True,
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def decode_type24(payload: str) -> dict | None:
|
|
"""Type 24 — Class B static (Part A: name; Part B: dimensions+callsign)."""
|
|
try:
|
|
msg_type = _bits(payload, 0, 6)
|
|
if msg_type != 24:
|
|
return None
|
|
mmsi = _bits(payload, 8, 30)
|
|
part_no = _bits(payload, 38, 2)
|
|
|
|
if part_no == 0:
|
|
name = _text6(payload, 40, 120)
|
|
return {
|
|
"type": "vessel_static",
|
|
"mmsi": str(mmsi),
|
|
"nombre": name or None,
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"source": "AIS",
|
|
}
|
|
elif part_no == 1:
|
|
ship_type = _bits(payload, 40, 8)
|
|
callsign = _text6(payload, 90, 42)
|
|
to_bow = _bits(payload, 132, 9)
|
|
to_stern = _bits(payload, 141, 9)
|
|
to_port = _bits(payload, 150, 6)
|
|
to_stbd = _bits(payload, 156, 6)
|
|
length = to_bow + to_stern
|
|
beam = to_port + to_stbd
|
|
return {
|
|
"type": "vessel_static",
|
|
"mmsi": str(mmsi),
|
|
"callsign": callsign or None,
|
|
"tipo": ship_type,
|
|
"tipo_nombre": SHIP_TYPE_NAMES.get(ship_type, f"Type {ship_type}"),
|
|
"length": length if length else None,
|
|
"beam": beam if beam else None,
|
|
"to_bow": to_bow if to_bow else None,
|
|
"to_stern": to_stern if to_stern else None,
|
|
"to_port": to_port if to_port else None,
|
|
"to_starboard": to_stbd if to_stbd else None,
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"source": "AIS",
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def parse_nmea_sentence(sentence: str) -> dict | None:
|
|
"""
|
|
Parse a single complete !AIVDM/!AIVDO NMEA sentence.
|
|
Returns decoded dict or None if unrecognised/invalid.
|
|
"""
|
|
sentence = sentence.strip()
|
|
if not (sentence.startswith("!AIVDM") or sentence.startswith("!AIVDO")):
|
|
return None
|
|
|
|
# Strip checksum
|
|
if '*' in sentence:
|
|
sentence = sentence[:sentence.rindex('*')]
|
|
|
|
parts = sentence.split(',')
|
|
if len(parts) < 6:
|
|
return None
|
|
|
|
payload = parts[5]
|
|
if not payload:
|
|
return None
|
|
|
|
try:
|
|
msg_type = _bits(payload, 0, 6)
|
|
except Exception:
|
|
return None
|
|
|
|
if msg_type in (1, 2, 3):
|
|
return decode_type123(payload)
|
|
elif msg_type == 5:
|
|
return decode_type5(payload)
|
|
elif msg_type == 18:
|
|
return decode_type18(payload)
|
|
elif msg_type == 24:
|
|
return decode_type24(payload)
|
|
elif msg_type in (21, 8):
|
|
# Handled by aton_decoder — return raw info so caller can route
|
|
return {"type": "aton_raw", "msg_type": msg_type, "payload": payload}
|
|
|
|
return None
|