ไล่ Signal Chain ที่ sensor ชายฝั่งทุกตัวต้องเดินผ่าน ตั้งแต่ตั้ง baud rate บนขา RS-232 ผ่าน serial device server จนแกะข้อความ AIS ขึ้นเป็นหมุดบนแผนที่ แบบไม่ต้องแต่งงานกับยี่ห้อไหน
สวัสดีครับวัยรุ่น ตอนที่แล้วผมเล่าภาพใหญ่ไว้ที่ Maritime Domain Awareness (MDA) ตระหนักรู้ไปทำไม? ว่ามันมีสามขา คือ Situational / Threat / Response แล้วขาแรกที่ชื่อ Maritime Situational Awareness มันแปลว่า "ตอนนี้ในทะเลมีอะไรอยู่บ้างวะ" — ฟังดูง่ายเนอะ แต่พอลงมือทำจริง มันคือการไล่ต่อสายทีละเส้น ตั้งกำหนด baud rate ทีละตัว แล้วนั่งแกะไบต์ทีละก้อน กว่าจะได้หมุดหนึ่งหมุดขึ้นจอ
ซีรีส์นี้ผมจะขยายขาแรกแบบ geek ยาวๆ เป็นตอนๆ และ สามตอนแรกขอคุยเฉพาะ sensor ในภาวะปกติ ที่จับเรือปกติซึ่งยอมเปิดตัวเอง ศัพท์ทางการเขาเรียก cooperative คือเรือมันบอกเราเอง หรือไม่ก็ยืนอยู่เฉยๆ ให้เราส่องได้โดยไม่หลบ ส่วนพวกที่ตั้งใจหายไปจากจอ (dark activities) เก็บไว้ตอนที่ 4
ที่ผมจะเล่าไม่ใช่นิยามนะครับ นิยามเปิด Wikipedia เอาก็ได้ สิ่งที่หาอ่านยากกว่าคือ "แล้วสายมันเสียบตรงไหน" เพราะฉะนั้นแต่ละ sensor ผมจะไล่ให้ครบห้าจังหวะเหมือนกันหมด คือ ต่อสายยังไง → config อะไรบ้าง → ข้อความหน้าตาแบบไหน → parse ด้วยโค้ดยังไง → แล้วเอาขึ้นแผนที่ให้คนดูรู้เรื่องยังไง (โค้ดตัวอย่างทั้งซีรีส์เป็น Python ล้วน เพราะงานฝั่งดูดสัญญาณกับแกะไบต์มันลงเอยที่ Python แทบทุกที่)
ทีแรกผมเขียนรวดเดียวจบ แล้วอ่านทวนอีกทีถึงรู้ว่ามันยาวจนคนอ่านนิ้วล้าก่อนถึงครึ่งเรื่อง เลยขอซอยเป็นสามตอน — ตอนที่ 1 นี้ว่าด้วยจุดยืนเรื่องมาตรฐาน Signal Chain ที่ทุก sensor ต้องเดินผ่าน และ AIS receiver ชายฝั่ง ตอนที่ 2 เป็นเรดาร์ชายฝั่งกับกล้อง EO/IR ส่วนตอนที่ 3 เอาเป้าจาก AIS เรดาร์ และกล้อง ไปพล็อตลงบนแผนที่ GIS แล้วปิดด้วยเรื่องที่มันส์ที่สุด คือการ Fusion และ Correlation เป้า ให้เรือหนึ่งลำเหลือหมุดเดียว ไม่ใช่สี่หมุดกองกันอยู่
ขอเคลียร์จุดยืนก่อนเลย เพราะนี่คือแผลเป็นที่คนทำระบบสาย sensor มีเหมือนกันหมด — เวลาซื้ออุปกรณ์ คนขายจะพยายามขาย protocol ของตัวเองพ่วงมาด้วยเสมอ
มันจะมาในสำนวนประมาณว่า "ของเราเอาต์พุตเป็นมาตรฐานก็ได้ครับ แต่ถ้าใช้ SDK ของเราจะได้ข้อมูลครบกว่านะ" หรือ "ตัวนี้มีโปรแกรมเฝ้าระวังของเราให้ฟรี ต่อ API ของเราง่ายกว่าเยอะ" ฟังดูใจดีมาก จนกระทั่งวันที่คุณอยากเปลี่ยนยี่ห้อ หรือวันที่เขาขึ้นราคา license แล้วคุณพบว่าระบบทั้งระบบมันผูกกับ DLL ตัวหนึ่งที่คุณไม่มีซอร์ส และคนเขียนลาออกไปสามปีแล้ว
กฎที่ผมยึดคือ สัญญาต้องอยู่ที่มาตรฐาน ไม่ใช่ที่ยี่ห้อ ตอนเขียน TOR หรือตอนคุยสเปก ประโยคสำคัญที่สุดคือ "อุปกรณ์ต้องเอาต์พุตตามมาตรฐาน X ทางพอร์ตกายภาพ Y โดยไม่ต้องผ่านซอฟต์แวร์ของผู้ผลิต" แค่ประโยคเดียวนี้ช่วยชีวิตระบบไว้ได้หลายปี
ในโค้ดก็คิดแบบเดียวกัน ทุก sensor ที่ต่อเข้ามาจะถูกบีบให้พูดภาษาเดียวกันตั้งแต่ประตูหน้า ผมเรียกมันว่า adapter ตัวหนึ่งต่อหนึ่งชนิดสัญญาณ แล้วข้างในระบบรู้จักแค่ก้อนเดียวคือ Track
from dataclasses import dataclass, field
from typing import Literal, Optional
@dataclass
class Track:
id: str # id ภายในของเรา ไม่ใช่เลขของ sensor
source: tuple[Literal["ais", "radar", "eoir", "vms"], str] # (ชนิด sensor, รหัสสถานี)
t: float # epoch ms — UTC เท่านั้น ห้ามมีเวลาไทยหลุดเข้ามา
lat: float # WGS84 เท่านั้น
lon: float
sog: Optional[float] = None # knots
cog: Optional[float] = None # องศา true
heading: Optional[float] = None
identity: dict = field(default_factory=dict) # mmsi / name / callsign
pos_acc_m: Optional[float] = None
latency_ms: float = 0.0
raw: Optional[str] = None # เก็บของดิบไว้ด้วย วันเถียงกันจะได้มีหลักฐานฟิลด์ raw นี่สำคัญกว่าที่คิดนะครับ เพราะวันที่ operator โวยว่า "เป้านี้มันผิด" สิ่งที่ตัดจบได้คือข้อความดิบกับ timestamp ตอนรับ ไม่ใช่ความทรงจำของเรา
ข้อสอบข้อเดียวที่ใช้วัดว่าออกแบบถูกไหม: ถอด sensor ตัวไหนออกก็ได้ แล้วที่เหลือยังวิ่งต่อเหมือนไม่มีอะไรเกิดขึ้น ถ้าถอดแล้วจอดับทั้งจอ แปลว่าเราไม่ได้ทำระบบ เราแค่ทำ GUI ให้อุปกรณ์ตัวนั้น
ก่อนจะแยกไปทีละตัว มาดูของที่เหมือนกันหมดก่อน เพราะ 80% ของงาน sensor ชายฝั่งมันคือ Signal Chain เส้นนี้เส้นเดียว
จังหวะที่ 1 — พอร์ตกายภาพ อุปกรณ์เดินเรือส่วนใหญ่ยังคุยกันด้วย serial ครับ ไม่ใช่เพราะล้าหลัง แต่เพราะมันนิ่งและซ่อมง่าย ที่เจอบ่อยมีสองแบบ
ข้อนี้สำคัญ ถ้าเจ้าของอุปกรณ์มีจอของเขาใช้อยู่แล้ว อย่าไปดึงสายเขามาเสียบเรา ให้แยกสัญญาณด้วยตัวแยกที่มี opto-isolation (พวกที่เรียกกันว่า NMEA multiplexer) หรือใช้คุณสมบัติ multi-listener ของ RS-422 ไม่งั้นวันดีคืนดีจอเขาดับ เราคือจำเลย
จังหวะที่ 2 — ตั้งค่าพอร์ต ตรงนี้คือจุดที่มือใหม่เสียเวลาเป็นวัน ทั้งที่มันมีแค่สี่ค่า: baud rate, data bits, parity, stop bits เขียนย่อกันว่า 4800 8N1 = 4800 บิตต่อวินาที, ข้อมูล 8 บิต, ไม่มี parity, stop bit 1 ตัว ค่ามาตรฐานที่ควรจำ
ถ้าเปิดพอร์ตแล้วเจออักขระขยะเป็นพรืด ไม่ต้องตกใจว่าอุปกรณ์พัง 90% คือ baud rate ผิด ลองไล่ 4800 → 9600 → 38400 → 115200 อีก 9% คือสลับขา TX/RX (ต้องไขว้กัน) หรือไม่ได้ต่อกราวด์ร่วม ที่เหลือ 1% ค่อยว่ากัน
หน้าตาของ "ขยะจาก baud ผิด" เป็นประมาณนี้ครับ (ตัวอย่างประกอบ ไม่ใช่ของจริงจากที่ไหน) สมมติเปิดพอร์ตของเครื่องรับ AIS ที่ 4800 ทั้งที่ของจริงมันส่งมาที่ 38400
# (ก) ser = serial.Serial(port, 4800) — ผิด baud ไป 8 เท่า
>>> ser.read(48)
b'\xf8\xe0\x80\x00\xfe\xc1\xf0\x1c\x80\xff\x83\xe0\xf8\x07\xc0\xfe\x1f\x00\xe0\xf8\xc3\x80\x7f\xf0'
>>> ser.read(48)
b'\x00\xf8\xff\x07\xe0\x8f\xc0\xf8\x1f\x80\xfe\x00\xe3\xf8\x80\xff\xf0\x0f\xc0\xe0\xf8\x83\x00\xfe'
# (ข) ser = serial.Serial(port, 38400) — พอร์ตเดิม สายเส้นเดิม เปลี่ยนแค่เลขตัวเดียว
>>> ser.read(48)
b'!AIVDM,1,1,,B,139K?4?01;7<bb`7Bk2b5`4J0H88,0*2D\r\n'ดูสองอย่างพอครับ หนึ่ง มีตัวขึ้นต้นกับตัวจบที่คงที่ไหม ของจริงต้องขึ้นด้วย $ หรือ ! แล้วปิดด้วย * ตามด้วย checksum สองหลักและ \r\n เป็นจังหวะสม่ำเสมอ ส่วนขยะจาก baud ผิดจะไม่มีอะไรซ้ำเป็นแพตเทิร์นเลย สอง มี byte ค่าสูงๆ อย่าง \xff \xfe \xf8 \xe0 \x00 โผล่เยอะผิดปกติไหม NMEA 0183 เป็น ASCII ล้วน ถ้าเห็นพวกนี้แปลว่า UART กำลังอ่าน bit ผิดกรอบ ไม่ใช่ข้อมูลพัง
ที่ต้องแยกให้ออกคือ "ขยะออกมา" กับ "ไม่มีอะไรออกมาเลย" เป็นคนละอาการ ขยะแปลว่าสัญญาณมาถึงแล้วแต่ตั้งค่าผิด — ไล่ baud rate ต่อได้เลย ส่วนเงียบสนิทอ่านได้แต่ b'' มักเป็นเรื่องสายล้วนๆ คือ TX/RX ไม่ได้ไขว้กัน หรือกราวด์ไม่ถึง รู้แค่นี้ก็ตัดทางเดาไปได้ครึ่งหนึ่งแล้ว
จังหวะที่ 3 — เอาขึ้นเน็ตเวิร์ก เราไม่ลากสาย serial ยาวๆ กลับห้อง server หรอกครับ เราเอา serial device server (บางทีเรียก terminal server) ไปวางใกล้อุปกรณ์ แล้วให้มันแปลง serial เป็น Ethernet โหมดที่ใช้บ่อย
จังหวะที่ 4 — ฝั่ง server ต้องตัดบรรทัดให้เป็น ข้อผิดพลาดคลาสสิกของคนเพิ่งทำ streaming คือนึกว่า TCP หนึ่ง packet เท่ากับหนึ่งข้อความ ไม่จริงครับ มันเป็น stream ของไบต์ล้วนๆ ข้อความเดียวอาจมาสามก้อน หรือสามข้อความอาจมาก้อนเดียว หน้าที่เราคือ buffer แล้วตัดตาม \r\n เอง
# tap.py — ดูด NMEA 0183 ได้ทั้งเสียบพอร์ตตรงๆ และรับผ่าน serial device server
import math, socket, time
import serial # pyserial
def serial_source(port="/dev/ttyUSB0", baud=38400):
# 8N1 เป็นค่าปริยายของ pyserial อยู่แล้ว เขียนไว้ให้เห็นกับตา
with serial.Serial(port, baud, bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE,
timeout=1) as ser:
while True:
chunk = ser.read(4096)
if chunk:
yield chunk
def tcp_source(host="10.10.20.31", port=4001): # พอร์ตที่ผูกกับ serial port ของ sensor ตัวนั้น
while True: # สายหลุดเมื่อไหร่ต่อใหม่เอง อย่ารอคนมากด
try:
with socket.create_connection((host, port), timeout=10) as s:
while True:
chunk = s.recv(4096)
if not chunk:
break # ปลายทางปิด = ออกไป reconnect
yield chunk
except OSError as e:
print("[tap]", e)
time.sleep(3)
def udp_source(port=4002): # กรณีกล่องยิง broadcast/multicast มาให้
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("", port))
while True:
yield s.recv(4096)
def lines(source):
buf = b""
for chunk in source:
buf += chunk
while b"\r\n" in buf:
raw, buf = buf.split(b"\r\n", 1)
yield raw.decode("ascii", errors="ignore").strip()
if len(buf) > 4096:
buf = b"" # กันขยะค้าง buffer จนบวมตาย
def checksum_ok(s):
# NMEA 0183: XOR ของทุกตัวอักษรระหว่าง $ (หรือ !) กับ *
star = s.rfind("*")
if star < 1 or len(s) - star < 3:
return False
x = 0
for ch in s[1:star]:
x ^= ord(ch)
return x == int(s[star + 1:star + 3], 16)
for line in lines(tcp_source()):
if line and checksum_ok(line):
handle(line)ท่องไว้สามข้อ: ต่อใหม่เองได้ ตัดบรรทัดเอง เช็ก checksum ทุกบรรทัด เพราะของจริงมีบรรทัดพิการปนมาเสมอ ทั้งจากสัญญาณรบกวนและจากอุปกรณ์ที่ reboot กลางคัน (รายละเอียดมาตรฐานอ่านต่อที่ NMEA 0183)
เริ่มจากตัวที่ง่ายที่สุดก่อน สถานีรับ AIS ชายฝั่งมีของแค่สามชิ้น: เสาอากาศ VHF ที่ยิ่งสูงยิ่งเห็นไกล, เครื่องรับ AIS ที่เฝ้าสองช่องสัญญาณย่านทางทะเล (161.975 กับ 162.025 MHz) และสายที่ลากลงมาจากเครื่องรับ
จุดที่ผมชอบคือ AIS แทบไม่ต้อง config อะไรเลย เพราะเครื่องรับมันไม่ได้ส่งอะไรออกอากาศ (ถ้าเป็นตัวรับอย่างเดียว) มันแค่นั่งฟังแล้วพ่นข้อความออกพอร์ต 38400 8N1 เสียบปุ๊บติดปั๊บ ของที่ออกมาหน้าตาแบบนี้
!AIVDM,1,1,,B,139K?4?01;7<bb`7Bk2b5`4J0H88,0*2D
!AIVDM,2,1,3,B,55P5TL01VIaAL@7WKO@mBplU@<PDhh00000000,0*70
!AIVDM,2,2,3,B,1@0000000000000,2*55
\s:STN001,c:1789120800*48\!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23บรรทัดแรกคือข้อความสั้นที่จบในตัวเดียว บรรทัดที่สองกับสามคือข้อความยาวที่ถูกหั่นเป็นสองชิ้น ส่วนบรรทัดสุดท้ายมีของแถมข้างหน้าที่คั่นด้วย \ เรียกว่า tag block ใช้บอกว่าข้อความนี้มาจากสถานีไหน (s:) และรับเมื่อไหร่ (c: เป็น epoch วินาที) — ถ้าคุณมีสถานีรับหลายจุดแล้วรวมสายมาที่เดียว เจ้า tag block นี่แหละคือสิ่งที่ทำให้ยังแยกออกว่าใครได้ยินอะไร และใช้คำนวณ latency ได้ด้วย
ส่วนการถอด payload ตรงกลางที่ดูเหมือนแมวเดินบนคีย์บอร์ด ผมเล่าละเอียดยันทีละบิตไว้แล้วที่ Maritime Sensors Integration ไม่เล่าซ้ำ ตอนนี้ขอเจาะเฉพาะจุดที่คนพลาดกันเยอะ คือ การประกอบชิ้นส่วนก่อนถอด
import time
pending = {} # (ช่อง, seq, จำนวนชิ้น) -> ชิ้นส่วนที่รอเพื่อน
def on_ais_sentence(line):
f = line.split(",")
total = int(f[1]) # จำนวนชิ้นทั้งหมด
idx = int(f[2]) # ชิ้นที่เท่าไหร่ เริ่มนับที่ 1
seq = f[3] or "-" # sequential message id ใช้จับคู่ชิ้นส่วน
ch = f[4] # ช่อง A หรือ B
payload = f[5]
fill = int(f[6].split("*")[0]) # fill bits ของชิ้นสุดท้าย
if total == 1:
return decode_payload(payload, fill, ch)
key = (ch, seq, total)
e = pending.setdefault(key, {"parts": [None] * total, "got": 0, "ts": time.time()})
if e["parts"][idx - 1] is None:
e["got"] += 1
e["parts"][idx - 1] = payload
if e["got"] == total:
del pending[key]
decode_payload("".join(e["parts"]), fill, ch) # นับ fill bits จากชิ้นสุดท้ายเท่านั้น
def sweep_pending(ttl=10.0):
# เรียกเป็นระยะ — ชิ้นส่วนที่รอเกิน 10 วิแล้วเพื่อนไม่มา = ทิ้ง ไม่งั้น dict บวมจนตาย
now = time.time()
for k in [k for k, e in pending.items() if now - e["ts"] > ttl]:
del pending[k]ที่ต้องระวังคือ sequential message id มันวนแค่ 0–9 ครับ ถ้ามีชิ้นส่วนค้างแล้วเลขวนกลับมาซ้ำ คุณจะได้ข้อความลูกผสมที่ถอดออกมาเป็นเรือผีทันที เพราะงั้น timeout กับการเคลียร์ของค้างไม่ใช่ของหรูหรา มันคือของบังคับ
เอาขึ้นแผนที่ AIS ให้ของครบที่สุดในบรรดา sensor ทั้งหมด เพราะมันมีทั้งตำแหน่ง เข็ม ความเร็ว และ "ชื่อ" ติดมาด้วย แต่ต้องจำไว้ว่าข้อมูลมันมาคนละข้อความ ตำแหน่งมาทุกไม่กี่วินาที ส่วนชื่อเรือกับขนาดเรืออยู่ในข้อความ static ที่มาทุกไม่กี่นาที ดังนั้นฝั่ง state ต้องแยกเป็นสองตาราง แล้วค่อย join ด้วย MMSI ตอนแสดงผล
บนจอผมแนะนำแบบนี้ครับ
ไล่มาถึงตรงนี้เราได้ของสองชิ้น คือ Signal Chain ที่ทุก sensor ต้องเดินผ่านเหมือนกันหมด (พอร์ตกายภาพ → ตั้งค่าพอร์ต → serial device server → ตัดบรรทัดให้เป็นฝั่ง server) กับ AIS ที่เสียบปุ๊บติดปั๊บและพูดเก่งที่สุดในกลุ่มเพื่อน
ตอนที่ 2 จะไปต่อที่ Coastal radar อุปกรณ์ตัวเดียวที่พ่นข้อมูลออกมาได้ถึงสาม Message Format คือ NMEA 0183, NMEA 2000 และ ASTERIX CAT048 ซึ่งแต่ละ Message Format คนละโลกกันเลย แล้วปิดตอนด้วยกล้อง EO/IR กับการ slew-to-cue คลิกเป้าบนแผนที่แล้วกล้องหมุนไปเอง — ตัวนี้สนุกตรงที่คณิตศาสตร์ต้องถูกก่อน กล้องถึงจะหันถูก
ระหว่างนี้ใครเคยจิ้มสาย serial แล้วเจอกับดักแปลกๆ มาเล่าให้ฟังบ้าง โดยเฉพาะพวกที่เอกสารบอกอย่างแต่ของจริงเป็นอีกอย่าง แล้วเจอกันตอนหน้าครับวัยรุ่น
ตามหาเรือที่ดับ AIS ด้วย RF Detection ชายฝั่ง RF จากดาวเทียม ดาวเทียม SAR และ EO/IR แล้วให้ Machine Learning ชี้เป้าบนภาพ ก่อนเอา detection ที่ไม่มีชื่อกลับมาจับคู่กับ track ที่หลอมไว้ตอนที่แล้ว
หลอมเป้าจาก AIS เรดาร์ และกล้อง ให้เรือลำเดียวที่โผล่สี่หมุดเหลือหมุดเดียว ตั้งแต่ gating กับ association ไปจนถึงการหั่น window ตอนเป้าเป็นพัน แล้ววาด tiled map กับเป้าทั้งกองเองด้วย WebGPU
แกะทางออกข้อมูลสามแบบของเรดาร์ชายฝั่ง ทั้ง NMEA 0183 NMEA 2000 และ ASTERIX CAT048 แล้วต่อด้วยกล้อง EO/IR ที่สั่งผ่าน ONVIF ให้คลิกเป้าบนแผนที่แล้วหมุนไปเอง