ข้อมูลเต็มโกดังแล้วอย่าเพิ่งโยนให้ AI — ดูวิธีที่ออสเตรเลียบีบภัยคุกคามทางทะเลให้เหลือ 8 ตัวแล้วโฟกัสตามพื้นที่กับเวลา ก่อนแปลงความรู้ของนักวิเคราะห์เป็น rule, risk score และการแจ้งเตือนด้วย Python
ใครตามมาตั้งแต่ Maritime Domain Awareness คงจำได้ว่าเลาแบ่ง MDA ไว้สามขา คือ Situational (ตอนนี้ในทะเลมีอะไร), Threat (อันไหนน่าจะมีพิษ) และ Response (ถ้ามีพิษจริง เรามีอะไรส่งไปได้) แล้วขาแรกขาเดียวก็ลากยาวไปหกตอน ตั้งแต่ Signal Chain กับ AIS ในตอนที่ 1 ไปจนถึง การแลกข้อมูลกับเพื่อนต่างประเทศในตอนที่ 6
หกตอนนั้นทำงานอย่างเดียวกันหมด คือ เก็บ — เก็บ track ที่หลอมแล้ว เก็บ detection จากดาวเทียม เก็บรายงานภาคสนาม เก็บของที่เพื่อนบ้านส่งมาให้ ทุกอย่างไหลลง Data Lake พร้อมที่มากำกับครบ ข้อมูลเต็มโกดังแล้ว คำถามต่อมาก็คือ แล้วไงต่อวะ
ตอนที่ 5 ผมทิ้งท้ายไว้ว่า Threat Awareness ไม่ใช่กล่องดำที่พ่นคำตอบว่าเรือลำนี้ผิด แต่คือการ จัดคิว ว่าเรือตรวจการณ์ที่ว่างอยู่ลำเดียวควรไปหาใครก่อน บทความนี้จะแกะว่าคิวนั้นมาจากไหน ใครเป็นคนตั้ง แล้วเปลี่ยนเป็นโค้ดได้ยังไง ระหว่างทางจะหยิบของที่เคยเล่าไว้แบบเร็วๆ ใน Behavior Analysis มาขยายให้ละเอียดขึ้นด้วย
ประโยคยอดฮิตเวลาระบบเริ่มมีข้อมูลเยอะคือ "เดี๋ยวโยนให้ AI ดู" ฟังดูทันสมัยดี แต่ส่วนใหญ่พูดกันตอนที่ยังไม่มีใครนั่งวิเคราะห์ข้อมูลกองนั้นจริงจังสักรอบ ยังไม่เข้าใจว่าเรือแต่ละแบบปกติทำตัวยังไง ยังไม่เข้าใจว่า dark activities ในพื้นที่ของเราหน้าตาแบบไหน แล้วก็หวังให้โมเดลไปหาคำตอบเอง
ความจริงขั้นนี้ใช้ คน เป็นหลัก คนที่รู้จักทะเลตรงนั้น รู้ว่าเรือประมงออกกี่โมง ฤดูไหนปลาเข้า ร่องน้ำไหนเรือสินค้าชอบตัด แล้วเอาความรู้นั้นมาเขียนเป็น Rule ให้ Machine ทำแทนตาคนต่อ Machine เก่งเรื่องดูทั้งวันทั้งคืนโดยไม่เบื่อ แต่ถ้าไม่มีใครบอก มันก็ไม่รู้ว่าจะดูอะไร
ถ้าไปเปิดคู่มือนักวิเคราะห์ข่าวกรองของ UNODC จะเจอวงรอบที่ละเอียดกว่าวงของป้าในบทความ MDA คือ Tasking → Collection → Evaluation → Collation → Analysis → Inference → Dissemination กับประโยคที่ผมชอบมาก information + evaluation = intelligence ข้อมูลดิบที่ยังไม่ผ่านการประเมินยังไม่นับเป็นข่าวกรอง ส่วนการโยนให้ AI ตั้งแต่วันแรกก็คือการกระโดดจาก Collection ไป Dissemination ข้ามทุกขั้นตรงกลางไปหน้าตาเฉย
ฝั่งงานวิจัยเองก็ไปทางเดียวกัน วิธีตรวจจับพฤติกรรมเรือส่วนใหญ่ยังต้องเริ่มจากนิยาม query และ rule ของแต่ละพฤติกรรมก่อน สายนี้เรียกว่า knowledge-driven ส่วนสาย data-driven ที่ให้โมเดลเรียนเองจากข้อมูล ก็ยังต้องมีคนมาบอกว่าเคสไหนผิดปกติจริง ถึงจะมี label ให้มันเรียน สรุปคือหนีคนไม่พ้นทั้งสองทาง
อีกเรื่องที่ต้องแยกให้ออกก่อนไปต่อ threat กับ risk ไม่ใช่คำเดียวกัน เอกสารของ BIMCO (สมาคมเจ้าของเรือระดับโลก) เขียนแยกไว้ชัด
เรือประมงเถื่อนหนึ่งลำคือ threat ส่วนคำถามว่า "เดือนหน้าทะเลช่องนี้น่ากลัวแค่ไหน" คือ risk ระบบที่จะเขียนต่อจากนี้ใช้ทั้งคู่ แต่ถ้าเอามาปนกันมั่ว จะได้ตัวเลขที่ไม่มีใครรู้ว่ามันวัดอะไรอยู่
ลองลิสต์ภัยคุกคามทางทะเลเล่นๆ ดู ประมงเถื่อน ขนคน ขนยา ถ่ายน้ำมันเถื่อนกลางทะเล โจรสลัด เรือลากสมอทับสายเคเบิล ทิ้งขยะ แมลงติดมากับสินค้า เรือสำรวจที่ไม่ได้ขออนุญาต … นั่งเขียนสักชั่วโมงได้เป็นร้อยข้อ แล้วแต่ละข้อยังแตกย่อยได้อีก
จุดที่หลายคนพลาดคือพยายามคิดให้ ครอบคลุม ไว้ก่อน ทุกภัยต้องมี ทุกพื้นที่ต้องดู ทุกช่วงเวลาต้องเฝ้า ผลคือได้ระบบที่ต้องดูทุกอย่าง แล้วก็โฟกัสอะไรไม่ได้สักอย่าง จอที่แดงทั้งจอมีค่าเท่ากับจอที่ไม่แดงเลย
ขั้น Threat Awareness คือหัวใจที่ทำให้เรารู้ว่าจะงมเข็มในมหาสมุทรตรงไหน วิธีคือบีบทีละชั้น
บีบครบสามชั้นจะเหลือภัยไม่กี่ตัวที่ต้องดู ดูได้จริง และคุ้มที่จะดู ในพื้นที่กับช่วงเวลานั้นๆ
ตัวอย่างที่เห็นภาพชัดคือออสเตรเลีย หน่วย Australian Border Force (ABF) บอกว่าต้องดูแลชายฝั่งยาว 37,000 กม. กับพื้นที่ทางทะเลราว 10% ของผิวโลก ส่วนรายงานตรวจสอบของ ANAO (สำนักงานตรวจเงินแผ่นดินของออสเตรเลีย) ก็เขียนตรงๆ ว่าเมื่อ asset มีจำกัดแต่พื้นที่มหาศาลขนาดนี้ ทางเดียวคือจัดลำดับด้วย risk management — ไม่มีใครมีเรือพอจะเฝ้าทุกตารางไมล์หรอก
ถ้าเขียนเป็นโค้ด ขั้นบีบนี้หน้าตาเรียบง่ายมาก ที่ยากคือการกรอกตาราง ซึ่งเป็นงานของคนล้วนๆ
from dataclasses import dataclass
@dataclass
class Threat:
name: str
zones: set # พื้นที่ที่ภัยนี้มีเหตุผลจะเกิด
months: set # เดือนที่ต้องจับตา (1-12)
needs: set # ชั้นข้อมูลที่ต้องมี ถึงจะ "เห็น" ภัยนี้ได้จริง
# รายการตัดมาแค่บางส่วน ของจริงยาวกว่านี้เยอะ — ทุกค่าในตารางนี้เป็นค่าสมมติ
THREATS = [
Threat("ประมงผิดกฎหมาย", {"A", "B"}, {11, 12, 1, 2, 3}, {"ais", "sar", "fishing_zones"}),
Threat("ลักลอบขนคนทางเรือ", {"B", "C"}, {4, 5, 6, 7}, {"coastal_radar", "humint"}),
Threat("ลักลอบขนยาเสพติด", {"A", "C"}, set(range(1, 13)), {"ais", "port_calls", "humint"}),
Threat("ถ่ายน้ำมันเถื่อนกลางทะเล", {"C"}, set(range(1, 13)), {"ais", "sar"}),
Threat("ทิ้งสมอ/ลากอวนใกล้เคเบิล", {"A"}, set(range(1, 13)), {"ais", "cable_routes"}),
Threat("ทำผิดในเขตอนุรักษ์", {"B"}, {1, 2, 3, 4}, {"ais", "protected_areas"}),
Threat("แมลง/สัตว์ติดมากับเรือ", {"A", "B", "C"}, {9, 10, 11, 12, 1, 2, 3, 4}, {"port_calls", "cargo_manifest"}),
Threat("ทิ้งของเสียกลางทะเล", {"A", "B", "C"}, set(range(1, 13)), {"sar", "oil_spill_model"}),
Threat("เรือสำรวจไม่ได้รับอนุญาต", {"C"}, set(range(1, 13)), {"ais", "rf_detection"}),
]
def focus_threats(threats, zone, month, have):
"""กรองสามชั้น: พื้นที่ -> ช่วงเวลา -> ข้อมูลที่มีจริง แล้วรายงานจำนวนที่เหลือทุกชั้น"""
by_zone = [t for t in threats if zone in t.zones]
by_month = [t for t in by_zone if month in t.months]
focus = [t for t in by_month if t.needs <= have]
blind = {t.name: sorted(t.needs - have) for t in by_month if not t.needs <= have}
funnel = f"{len(threats)} -> {len(by_zone)} -> {len(by_month)} -> {len(focus)}"
return funnel, [t.name for t in focus], blind
HAVE = {"ais", "sar", "coastal_radar", "fishing_zones", "protected_areas", "cable_routes", "port_calls"}
funnel, focus, blind = focus_threats(THREATS, zone="B", month=2, have=HAVE)
print(funnel) # 9 -> 5 -> 4 -> 2
print(focus) # ['ประมงผิดกฎหมาย', 'ทำผิดในเขตอนุรักษ์']
print(blind) # ภัยที่ควรดูแต่ตาบอด พร้อมบอกว่าขาดข้อมูลชั้นไหนผลที่ได้ 9 -> 5 -> 4 -> 2 คือจากเก้าภัยในรายการ เหลือสองภัยที่ต้องโฟกัสในโซน B เดือนกุมภาพันธ์ แต่ของที่มีค่าที่สุดกลับเป็นตัวแปร blind ที่บอกว่ามีภัยอีกสองตัวที่ ควรดูแต่ดูไม่ได้ เพราะขาดข้อมูลชั้นไหน อันนี้คือของที่ต้องหอบไปคุยกับคนถือเงินว่าควรลงทุนอะไรต่อ ดีกว่าแกล้งทำเป็นว่าเราเฝ้าครบทุกภัย
ออสเตรเลียไม่ได้เริ่มจากร้อยภัย รัฐบาลเขาบีบไว้ใน Civil Maritime Security Strategy ปี 2021 ให้เหลือ 8 ภัยคุกคามความมั่นคงทางทะเลฝั่งพลเรือน (ABF ก็สรุปไว้ในหน้าเว็บของตัวเอง)
คนที่รับรายการนี้ไปทำต่อคือ Maritime Border Command (MBC) ซึ่งเป็น taskforce รวมหลายหน่วย ทั้ง ABF, กองทัพ (ADF), ตำรวจสหพันธ์ (AFP) และหน่วยจัดการประมง (AFMA) โดยเป็นเจ้าภาพเรื่อง maritime domain awareness ของประเทศ แนวทางที่ประกาศไว้คือ "intelligence-informed, risk-based approach" แบ่งงานเป็น Deter / Detect / Respond ส่วนตัว Strategy วางกรอบใหญ่ไว้เป็น Understand / Influence / Prevent / Respond คร่าวๆ
ลองดูเคสช่วงปีที่ผ่านมา แล้วสังเกตว่าแต่ละเคสโฟกัสด้วยมิติไหน
สี่เคสนี้ไม่มีเคสไหนเกิดจากการนั่งจ้องเรือทุกลำในทุกจุดของทะเล ทุกเคสมีคนตัดสินใจไว้ก่อนแล้วว่าจะดูตรงไหน ตอนไหน และดูเรือแบบไหน
แล้วเขาตัดสินใจกันยังไงว่าจะไปดูตรงไหน อันนี้มีเอกสารเปิดที่เล่าละเอียดอยู่ฉบับหนึ่ง คือรายงาน ANAO Audit Report No.23 ปี 2009–10 ที่ตรวจ Border Protection Command ซึ่งเป็นรุ่นก่อนของ MBC ของเก่าหน่อย แต่ละเอียดกว่าที่อื่นเยอะ
มีรายละเอียดอีกสามข้อที่ผมว่าสำคัญกว่าตัวสูตร ข้อแรก ฤดูกาล ทั้งสภาพอากาศและแหล่งสัตว์น้ำเป็นตัวกำหนดความเสี่ยงของพื้นที่ ข้อสอง intelligence ส่วนใหญ่ ไม่ได้ชี้เรือรายลำ แต่ชี้ว่ากิจกรรมแบบไหนน่าจะเกิดในพื้นที่ไหน ข้อสาม แผนทำเป็นรายไตรมาส แล้วทบทวนรายเดือน รายสัปดาห์ จนถึงรายวัน ยิ่งกรอบเวลาสั้น intelligence ยิ่งเจาะจง
ส่วนคนวางแผนใช้อะไรตัดสิน รายงานเขียนไว้ว่า "expert knowledge of the operating environment, including established patterns of activity" — ความรู้ของคนที่รู้จักทะเลตรงนั้นกับรูปแบบกิจกรรมที่เห็นมาจนชิน วนกลับมาที่คนอีกแล้วไง
ฝั่ง AFMA เองก็เปิดวิธีประเมินความเสี่ยงด้าน compliance ช่วง 2023–25 ไว้ ใช้กรอบ ISO 31000 แยก inherent risk (consequence × likelihood ก่อนมีมาตรการใดๆ) ออกจาก residual risk (ที่เหลือหลังมี control แล้ว) ลงเป็น risk matrix สี และให้เจ้าหน้าที่ compliance กับ intelligence หลายคน ช่วยกันให้ความเห็น เพื่อกัน bias ของคนคนเดียว
heat map อันนี้คนละตัวกับที่เล่าไว้ในตอนที่ 5 นะฮะ อันนั้นนับเหตุการณ์ที่ เกิดไปแล้ว แล้วต้องหารด้วยความพยายามในการตรวจ ส่วน CRAM คือการ คาดการณ์ ล่วงหน้าว่าเดือนนี้ควรเอากำลังไปวางตรงไหน ใช้คู่กันได้ดี ของตอน 5 ใช้เป็นหลักฐานประกอบตอนให้คะแนน likelihood ของ CRAM
รายงานไม่ได้เปิดเผยน้ำหนักจริงหรือวิธีแบ่ง band แบบละเอียด โค้ดข้างล่างเลยเป็นแค่ท่าหนึ่งที่ทำตามหลักการเดียวกันได้ ตัวเลขทุกตัวเป็น ค่าสมมติ และผมแบ่ง band โดยเทียบกับช่องที่หนักที่สุดของเดือนนั้น
import math
# น้ำหนักผลกระทบสามด้าน — ค่าสมมติ รายงานไม่ได้เปิดเผยตัวเลขจริง
W = {"economic": 0.40, "environment": 0.35, "social": 0.25}
# consequence ของแต่ละภัย ให้คะแนน 1-5 ต่อด้าน (ค่าสมมติ)
CONSEQUENCE = {
"illegal_fishing": {"economic": 4, "environment": 5, "social": 2},
"prohibited_import": {"economic": 3, "environment": 1, "social": 5},
"biosecurity": {"economic": 5, "environment": 4, "social": 2},
}
def consequence(threat):
return sum(W[k] * v for k, v in CONSEQUENCE[threat].items()) # ได้ 1.0-5.0
def cram_month(cells, likelihood, month):
"""likelihood[(cell, threat)] = list 12 ค่า (1-5) ต่อเดือน มาจากนักวิเคราะห์ + intelligence outlook
คืน list ของ (cell, คะแนนรวม, band 1-10, ภัยตัวที่ดันคะแนน) เรียงจากเสี่ยงสุด"""
rows = []
for cell in cells:
per_threat = {t: consequence(t) * likelihood.get((cell, t), [0] * 12)[month - 1]
for t in CONSEQUENCE}
driver = max(per_threat, key=per_threat.get) # ภัยตัวที่ดันคะแนนช่องนี้
rows.append([cell, round(sum(per_threat.values()), 1), 0, driver])
top = max(r[1] for r in rows) or 1
for r in rows:
r[2] = max(1, math.ceil(r[1] / top * 10)) # band ช่วงละ 10% เทียบช่องที่หนักสุด
return sorted(map(tuple, rows), key=lambda r: r[1], reverse=True)
# ลองสามช่อง ช่อง 07 ประมงหนักช่วงปลายปี ช่อง 31 ติดเส้นทางเรือสินค้า — ค่าสมมติทั้งหมด
LIKELIHOOD = {
("07", "illegal_fishing"): [2, 2, 1, 1, 1, 1, 1, 2, 3, 4, 5, 5],
("07", "biosecurity"): [1] * 12,
("31", "prohibited_import"): [3] * 12,
("31", "biosecurity"): [4, 4, 4, 4, 1, 1, 1, 1, 4, 4, 4, 4],
("52", "illegal_fishing"): [1] * 12,
}
for month in (3, 11):
print(month, cram_month(["07", "31", "52"], LIKELIHOOD, month))
# 3 [('31', 24.0, 10, 'biosecurity'), ('07', 7.8, 4, 'biosecurity'), ('52', 3.9, 2, 'illegal_fishing')]
# 11 [('31', 24.0, 10, 'biosecurity'), ('07', 23.1, 10, 'illegal_fishing'), ('52', 3.9, 2, 'illegal_fishing')]ดูช่อง 07 ดีๆ เดือนมีนาคมอยู่ band 4 พอถึงพฤศจิกายนที่เป็นฤดูปลา (สมมติ) กระโดดไป band 10 ทั้งที่ไม่มีอะไรเปลี่ยนนอกจากปฏิทิน นี่คือ "ฤดูกาลกำหนดความเสี่ยงของพื้นที่" ในรูปโค้ด ส่วนคอลัมน์สุดท้ายที่บอกว่าภัยตัวไหนดันคะแนน ช่วยให้คนวางแผนรู้ว่าจะส่งอะไรไป ประมงก็เรือประมง biosecurity ก็ทีมตรวจที่ท่า คนละงานกันชัดๆ
CRAM ตอบว่าดูตรงไหน ตอนไหน ขั้นต่อไปคือ ดูอะไร ในเรือแต่ละลำ ซึ่งก็คือเรื่องที่เลาเคยเล่าไว้ใน Behavior Analysis ตอนนั้นเราทำ buffer รอบสายเคเบิล แล้วเช็คว่าเรือที่อยู่ในโซนวิ่งช้ากว่า 2 นอตไหม เป็นการเช็คทีละจุด ใช้บอกหลักการได้ แต่ของจริงต้องอัปเกรดสามเรื่อง
หน้าตาของ dark activities แต่ละแบบเล่าไว้แล้วในตอนที่ 4 รอบนี้จะเขียน rule ให้มันจริงๆ โดยยืมนิยามของ Global Fishing Watch ที่เปิดเป็นค่าสาธารณะ
GFW เองย้ำว่า encounter เป็นแค่ "indicator for review" และ loitering ก็อาจบริสุทธิ์ใจได้ เช่น เรือเสียรอซ่อม หรือรอคิวเข้าเทียบท่า ผลลัพธ์ของ rule จึงเป็น event ให้คนไปดูต่อ ไม่ใช่คำพิพากษา
เริ่มจากชั้นพื้นที่ก่อน ฟังก์ชัน context() ตอบว่าจุดนี้อยู่ในโซนไหน ห่างฝั่งเท่าไหร่ ห่างที่ทอดสมอเท่าไหร่ layer ทั้งหมดเป็นของสมมติกลางมหาสมุทรอินเดีย
from pyproj import Transformer
from shapely.geometry import Point, LineString, Polygon
from shapely.ops import transform
NM = 1852.0 # 1 ไมล์ทะเล = 1,852 เมตร
LAT0, LON0 = -2.0, 88.0 # จุดกลางพื้นที่สมมติกลางมหาสมุทรอินเดีย
# ฉายจาก WGS84 ไประนาบ local หน่วยเมตร รอบจุดกลางพื้นที่ (Azimuthal Equidistant)
# always_xy=True = รับ (lon, lat) เสมอ ลืมบรรทัดนี้ทีไร แกน x/y สลับกันทุกที
_fwd = Transformer.from_crs("EPSG:4326",
f"+proj=aeqd +lat_0={LAT0} +lon_0={LON0} +units=m +datum=WGS84",
always_xy=True)
def to_m(geom):
return transform(_fwd.transform, geom)
def xy(lat, lon):
return _fwd.transform(lon, lat)
# layer ทั้งหมดเป็นของสมมติ ของจริงโหลดจาก shapefile/GeoJSON ของแต่ละหน่วยงาน
SHORE = to_m(Polygon([(87.90, -2.10), (88.10, -2.10), (88.10, -1.90), (87.90, -1.90)])) # เกาะสมมติ
ANCHORAGE = [to_m(Point(88.12, -2.00))]
ZONES = {
"protected": to_m(Polygon([(88.30, -2.40), (88.60, -2.40), (88.60, -2.10), (88.30, -2.10)])),
"cable": to_m(LineString([(87.50, -2.60), (88.20, -2.30), (89.00, -2.30)])).buffer(1 * NM), # buffer 1 nm ค่าสมมติ
}
def context(lat, lon):
"""บอกว่าจุดนี้อยู่ในโซนไหนบ้าง ห่างฝั่งกี่ nm ห่างที่ทอดสมอกี่ km"""
p = Point(xy(lat, lon))
return {
"zones": [name for name, poly in ZONES.items() if poly.contains(p)],
"shore_nm": p.distance(SHORE) / NM,
"anchorage_km": min(p.distance(a) for a in ANCHORAGE) / 1000,
}
print(context(-2.25, 88.45)) # อยู่ในเขตอนุรักษ์
print(context(-2.30, 88.60)) # ทับแนวเคเบิล ห่างฝั่ง ~32 nmใครจำได้ ใน Behavior Analysis ผมบ่นไว้ว่านักเดินเรืออ่าน Lat/Lon แต่ตอนคำนวณกลับต้องใส่ Lon/Lat ใน pyproj ก็เรื่องเดียวกันเป๊ะ always_xy=True คือบรรทัดที่ช่วยชีวิตไว้ ส่วน projection แบบ Azimuthal Equidistant ที่ตั้งจุดกลางไว้กลางพื้นที่ ให้ระยะจากจุดกลางออกไปถูกต้องเป็นเมตร พื้นที่ขนาดไม่กี่ร้อยกิโลเมตรแบบนี้ใช้ได้สบาย ถ้าพื้นที่กว้างเป็นพันกิโลเมตรค่อยแบ่ง projection เป็นโซนๆ
ต่อด้วย loitering เดินตาม track ไปทีละจุด ขยายหน้าต่างไปเรื่อยๆ ตราบที่ความเร็วเฉลี่ยยังต่ำ แล้วค่อยเช็คว่านานพอและไกลฝั่งพอไหม
from statistics import mean
# ค่าจากนิยาม loitering ของ Global Fishing Watch (ค่าสาธารณะ)
LOITER_MAX_KN = 2.0 # ความเร็วเฉลี่ยต่ำกว่า 2 นอต
LOITER_MIN_SHORE = 20.0 # ห่างฝั่งเฉลี่ยอย่างน้อย 20 nm
LOITER_MIN_HOURS = 1.0 # นานเกิน 1 ชั่วโมง
MAX_GAP_S = 2 * 3600 # ข้อมูลขาดเกิน 2 ชม. ตัดเป็นคนละช่วง — ค่าสมมติ
def detect_loitering(track):
"""track = list ของ dict {t, lat, lon, sog} เรียงตามเวลา (t เป็นวินาที)
คืน list ของช่วง loitering พร้อมบริบทว่าไปวนอยู่ในโซนไหน"""
events, i = [], 0
while i < len(track):
j = i
# ขยายหน้าต่างไปเรื่อยๆ ตราบที่ความเร็วเฉลี่ยยังต่ำ และข้อมูลไม่ขาดช่วง
while (j + 1 < len(track)
and track[j + 1]["t"] - track[j]["t"] <= MAX_GAP_S
and mean(p["sog"] for p in track[i:j + 2]) < LOITER_MAX_KN):
j += 1
seg = track[i:j + 1]
while len(seg) > 1 and seg[-1]["sog"] >= LOITER_MAX_KN: # ตัดหางที่เริ่มออกตัวแล้วทิ้ง
seg = seg[:-1]
hours = (seg[-1]["t"] - seg[0]["t"]) / 3600
if hours > LOITER_MIN_HOURS and mean(p["sog"] for p in seg) < LOITER_MAX_KN:
ctx = [context(p["lat"], p["lon"]) for p in seg]
shore = mean(c["shore_nm"] for c in ctx)
if shore >= LOITER_MIN_SHORE:
events.append({
"type": "loitering", "start": seg[0]["t"], "hours": round(hours, 1),
"shore_nm": round(shore, 1),
"zones": sorted({z for c in ctx for z in c["zones"]}),
})
i = j + 1
return events
# เรือวนในเขตอนุรักษ์ ~1 นอต 3 ชม. แล้วออกตัว 10 นอต:
# [{'type': 'loitering', 'start': 0, 'hours': 3.0, 'shore_nm': 22.9, 'zones': ['protected']}]สังเกตว่าเช็คระยะห่างฝั่ง หลัง เจอช่วงช้าแล้ว ไม่ใช่ก่อน เพราะ context() แพงกว่าการเฉลี่ยตัวเลขหลายเท่า ทำแค่กับช่วงที่ผ่านด่านแรกมาแล้วพอ แล้วก็ต้องตัดช่วงที่ข้อมูลขาดหายออกจากกัน ไม่งั้นเรือที่เงียบไปครึ่งวันจะกลายเป็นเรือที่ "loitering ครึ่งวัน" ไปซะงั้น ช่วงเงียบแบบนั้นควรส่งไปให้ detector ของ AIS gap ที่เล่าไว้ในตอนที่ 4 จัดการแทน
encounter ยากกว่าตรงที่ต้องเทียบเป็นคู่ ถ้ามีเรือหนึ่งหมื่นลำในภาพเดียว เทียบทุกคู่ตรงๆ ก็ราวห้าสิบล้านคู่ต่อหนึ่งช่วงเวลา ท่ามาตรฐานคือโยนเรือลง grid ขนาดเท่ากับระยะที่สนใจ แล้วเทียบเฉพาะช่องตัวเองกับแปดช่องรอบๆ
from collections import defaultdict
from statistics import median
import math
# ค่าจากนิยาม encounter ของ Global Fishing Watch (ค่าสาธารณะ)
ENC_MAX_M = 500 # สองลำห่างกันไม่ถึง 500 เมตร
ENC_MIN_HOURS = 2.0 # อยู่ด้วยกันอย่างน้อย 2 ชั่วโมง
ENC_MAX_KN = 2.0 # ความเร็ว median ต่ำกว่า 2 นอต
ENC_MIN_ANCH_KM = 10.0 # ห่างที่ทอดสมออย่างน้อย 10 กม.
STEP_S = 600 # resample ตำแหน่งทุก 10 นาทีก่อนเข้าฟังก์ชัน — ค่าสมมติ
def close_pairs(snapshot):
"""snapshot = list ของ (mmsi, x, y, sog) ณ เวลาเดียวกัน
ใช้ grid ขนาด 500 ม. ดูแค่ช่องตัวเองกับ 8 ช่องรอบๆ ไม่ต้องเทียบทุกคู่"""
grid = defaultdict(list)
for v in snapshot:
grid[(int(v[1] // ENC_MAX_M), int(v[2] // ENC_MAX_M))].append(v)
for (gx, gy), members in grid.items():
for dx in (-1, 0, 1):
for dy in (-1, 0, 1):
for a in members:
for b in grid.get((gx + dx, gy + dy), []):
if a[0] < b[0] and math.dist(a[1:3], b[1:3]) < ENC_MAX_M:
yield a, b
def detect_encounters(snapshots):
"""snapshots = dict {t: [(mmsi, lat, lon, sog), ...]} ที่ resample ทุก STEP_S วินาทีแล้ว"""
open_, done = {}, []
for t in sorted(snapshots):
pos = {m: (lat, lon) for m, lat, lon, _ in snapshots[t]}
snap = [(m, *xy(lat, lon), sog) for m, lat, lon, sog in snapshots[t]]
seen = set()
for a, b in close_pairs(snap):
key = (a[0], b[0])
seen.add(key)
e = open_.setdefault(key, {"start": t, "sogs": [], "anch": []})
e["end"] = t
e["sogs"] += [a[3], b[3]]
e["anch"].append(context(*pos[a[0]])["anchorage_km"])
for key in [k for k in open_ if k not in seen]: # คู่ไหนแยกกันแล้ว ปิดเคส
done.append((key, open_.pop(key)))
done += list(open_.items())
events = []
for (m1, m2), e in done:
hours = (e["end"] - e["start"] + STEP_S) / 3600
if (hours >= ENC_MIN_HOURS and median(e["sogs"]) < ENC_MAX_KN
and min(e["anch"]) >= ENC_MIN_ANCH_KM):
events.append({"type": "encounter", "vessels": (m1, m2),
"start": e["start"], "hours": round(hours, 1)})
return events
# สองลำจอดคู่กันห่าง ~130 ม. นาน 2.7 ชม. กลางทะเล + ลำที่สามวิ่งผ่าน + อีกคู่จอดข้างที่ทอดสมอ:
# [{'type': 'encounter', 'vessels': (111000001, 111000002), 'start': 0, 'hours': 2.7}]ตัวที่ทำให้ encounter หลอกไม่ค่อยได้คือเงื่อนไขเรื่องที่ทอดสมอ คู่ที่สองในตัวอย่างจอดคู่กันนานกว่าคู่แรกอีก แต่อยู่ข้างที่ทอดสมอ ซึ่งเป็นเรื่องปกติที่สุดในโลก เลยไม่ถูกนับ ตรงนี้แหละที่ความรู้เรื่องพื้นที่ของคนเข้ามามีผลตรงๆ ถ้า layer ที่ทอดสมอผิดหรือไม่ครบ rule ที่ดีแค่ไหนก็ยกมือมั่ว
แต่จะออกตัวไว้ก่อนเลยว่า ข้อมูลเหล่านี้กับการคำนวน ข้าเจ้าเองก็พึ่งเอไอเป็นส่วนใหญ่เหมือนกัน เพราะไม่ถนัดเรื่องสัตว์ทะเลกับพวกการอนุรักษ์เท่าไหร่ แต่ถามมันมาทำความเข้าใจ และถ่ายทอดออกมาเพื่อเก็บไว้เป็นความรู้ตัวเองด้วยหละ
ได้ event มาเต็มมือแล้ว ของที่ขาดคือ ใคร ซึ่งเป็นงานของ watchlist รายชื่อเรือที่น่าจับตาจากงานข่าว จากเคสเก่า หรือจากเพื่อนบ้านที่แชร์มา อย่างเคสโคเคนข้างบน จุดเริ่มคือเบาะแสเรื่องซื้อเรือด้วยเงินสด ไม่ใช่พฤติกรรมกลางทะเล ของแบบนี้แหละที่ควรไปนอนรออยู่ใน watchlist
ความยากของ watchlist คือตัวตนเรือมันลื่น MMSI เปลี่ยนได้ ชื่อเปลี่ยนได้ ธงก็เปลี่ยนได้ เหมือนเคส Andrey Dolgov ที่เล่าไว้ใน MDA ที่เปลี่ยนชื่อเปลี่ยนธงจนคนตามงง watchlist ที่ดีจึงต้องเก็บหลายกุญแจต่อหนึ่งลำ แล้ว normalize ชื่อก่อนเทียบ ไม่งั้น SEA STAR-1 กับ Sea Star 1 จะกลายเป็นคนละลำ
จากนั้นค่อยรวมทุกอย่างเป็นคะแนน หลักที่ผมยึดคือ คะแนนต้องมาพร้อมเหตุผลเสมอ คนที่เห็นเลข 87 บนจอแล้วไม่รู้ว่ามาจากไหน จะเลือกได้แค่สองทาง เชื่อทั้งดุ้น หรือไม่เชื่อสักนิด ซึ่งผิดทั้งคู่
import re
def norm(name):
"""ตัดช่องว่าง ขีด จุด ออก ให้ 'SEA STAR-1' กับ 'Sea Star 1' เป็นตัวเดียวกัน"""
return re.sub(r"[^A-Z0-9]", "", name.upper())
# watchlist เก็บได้หลายกุญแจต่อหนึ่งลำ เพราะ MMSI เปลี่ยนได้ ชื่อเปลี่ยนได้ — ตัวอย่างสมมติ
WATCHLIST = {
"mmsi:111000001": "เคยโดนตรวจพบถ่ายของกลางทะเล",
"imo:9000001": "งานข่าวแจ้งว่าซื้อเรือด้วยเงินสด",
f"name:{norm('Sea Star 1')}": "ชื่อเก่าของเรือที่เคยถูกยึด",
}
# น้ำหนักทุกตัวเป็นค่าสมมติ ของจริงนักวิเคราะห์ตั้งแล้วปรับจากผลตรวจ
WEIGHTS = {"loitering": 15, "encounter": 25, "ais_gap": 20,
"zone:protected": 20, "zone:cable": 25, "watchlist": 30}
def score_vessel(v, events, cram_band):
"""v = {mmsi, imo, names}, events = ผลจาก detector, cram_band = band 1-10 ของช่องที่เรืออยู่
คืน (คะแนน 0-100, เหตุผลทีละบรรทัด)"""
score, reasons = 0, []
def add(points, why):
nonlocal score
score += points
reasons.append(f"+{points} {why}")
for e in events:
add(WEIGHTS[e["type"]], f'{e["type"]} {e.get("hours", "?")} ชม.')
for z in e.get("zones", []):
add(WEIGHTS[f"zone:{z}"], f"เกิดในโซน {z}")
keys = [f"mmsi:{v['mmsi']}", f"imo:{v.get('imo')}"] + [f"name:{norm(n)}" for n in v.get("names", [])]
hit = next((k for k in keys if k in WATCHLIST), None)
if hit:
add(WEIGHTS["watchlist"], f"ติด watchlist ({hit}): {WATCHLIST[hit]}")
factor = 0.5 + cram_band / 10 # ช่องที่ CRAM เดือนนี้แดง คูณแรงกว่า
reasons.append(f"x{factor:.1f} บริบทพื้นที่และเดือน (CRAM band {cram_band})")
return min(100, round(score * factor)), reasons
vessel = {"mmsi": 111000009, "imo": 9111111, "names": ["SEA STAR-1", "OCEAN PRIDE"]}
events = [{"type": "loitering", "hours": 3.0, "zones": ["protected"]},
{"type": "encounter", "hours": 2.7}]
score, why = score_vessel(vessel, events, cram_band=9)
print(score) # 100 (ชนเพดาน)
print("\n".join(why))
print(score_vessel(vessel, events, cram_band=2)[0]) # 63 — เหตุการณ์เดิม แต่อยู่ในช่องที่เดือนนี้เย็นลองทาบกับสามเหลี่ยม threat ข้างบนดู watchlist ให้ข้อมูลฝั่ง intent กับ capability (เคยทำ มีของ มีเงิน) ส่วนโซนกับ CRAM band ให้ฝั่ง opportunity (อยู่ในที่ที่ทำได้ ในช่วงที่คุ้มจะทำ) event จาก rule เป็นตัวบอกว่าตอนนี้กำลังมีอะไรเกิดขึ้นจริง ครบสามขาเมื่อไหร่คะแนนก็พุ่ง
บรรทัดสุดท้ายของตัวอย่างสำคัญที่สุด เหตุการณ์ชุดเดียวกันทุกอย่าง แต่ถ้าเกิดในช่องที่เดือนนี้เย็น คะแนนลดจาก 100 เหลือ 63 นี่คือจุดที่ CRAM จากหัวข้อก่อนหน้าต่อเข้ากับ rule ระดับเรือ งานวางแผนระดับพื้นที่กับงานจับพฤติกรรมระดับลำ ไม่ได้แยกกันอยู่คนละโลก
ส่วนความมั่นใจ ท่าบ้านๆ ที่ใช้ได้ผลคือนับว่าเหตุผลแต่ละข้อมาจากแหล่งอิสระกี่แหล่ง loitering ที่เห็นจากทั้ง AIS และภาพ SAR ย่อมหนักแน่นกว่าที่เห็นจาก AIS อย่างเดียว อันนี้ต้องย้อนไปพึ่งของที่ย้ำไว้ตั้งแต่ตอนที่ 4 คือเก็บที่มาของทุกค่าไว้เสมอ
ระบบที่คำนวณเก่งแต่แจ้งเตือนห่วยคือระบบที่ไม่มีใครใช้ งานวิจัยที่ Lloyd's Register เผยแพร่ปี 2026 บอกว่าลูกเรือบนเรือเจอ alert เป็น หลายหมื่นครั้งต่อวัน และความเชื่อใจที่มีต่อระบบก็ลดลง อันนั้นเป็นฝั่งลูกเรือ ไม่ใช่นักวิเคราะห์ในศูนย์ แต่อาการเดียวกันเป๊ะ ปลุกคนบ่อยเกินจนเขาเลิกฟัง แล้ววันที่เป็นของจริงก็ไม่มีใครลุก
หลักที่ใช้มีไม่กี่ข้อ
from collections import defaultdict
# ภัยไหนส่งให้ใคร — ชื่อโต๊ะสมมติ ของจริงผูกกับหน่วยที่มีอำนาจสั่งการเรื่องนั้น
OWNERS = {"illegal_fishing": "fisheries_desk", "cable": "infrastructure_desk",
"smuggling": "customs_desk"}
DEFAULT_OWNER = "watch_officer"
LEVELS = [(70, "URGENT"), (40, "REVIEW"), (0, "DIGEST")] # เกณฑ์คะแนน — ค่าสมมติ
COOLDOWN_S = 6 * 3600 # เรื่องเดิม เตือนซ้ำได้ทุก 6 ชม. — ค่าสมมติ
class AlertRouter:
def __init__(self, send):
self.send = send # ฟังก์ชันส่งจริง จะเป็นจอ แชท หรือวิทยุก็ได้
self.last = {} # (mmsi, threat) -> (เวลา, คะแนน) ที่เตือนล่าสุด
self.digest = [] # ของคะแนนต่ำ รวมส่งเป็นรอบ ไม่ปลุกใคร
self.outcome = defaultdict(lambda: [0, 0]) # rule -> [ส่งไปตรวจ, เจอของจริง]
def route(self, mmsi, threat, score, reasons, now):
level = next(name for floor, name in LEVELS if score >= floor)
if level == "DIGEST":
self.digest.append((mmsi, threat, score))
return "digest"
prev = self.last.get((mmsi, threat))
if prev and now - prev[0] < COOLDOWN_S and score <= prev[1]:
return "suppressed" # เรื่องเดิม ไม่ได้หนักขึ้น ไม่ต้องปลุกซ้ำ
self.last[(mmsi, threat)] = (now, score)
owner = OWNERS.get(threat, DEFAULT_OWNER)
self.send(owner, f"[{level}] MMSI {mmsi} {threat} score {score}\n " + "\n ".join(reasons))
return owner
def feedback(self, rule, found):
"""ผลตรวจจากหน้างานไหลกลับมา — rule ไหนยกมือแล้วเจอจริงบ่อยแค่ไหน"""
self.outcome[rule][0] += 1
self.outcome[rule][1] += int(found)
def hit_rate(self):
return {r: round(hit / sent, 2) for r, (sent, hit) in self.outcome.items()}
router = AlertRouter(send=lambda who, msg: print(f"-> {who}\n{msg}"))
print(router.route(111000009, "illegal_fishing", 81, ["+15 loitering 3.0 ชม."], now=0))
print(router.route(111000009, "illegal_fishing", 78, ["+15 loitering 3.5 ชม."], now=1800))
print(router.route(111000042, "cable", 22, ["+25 เกิดในโซน cable"], now=1900))
router.feedback("loitering", found=False)
router.feedback("loitering", found=True)
router.feedback("encounter", found=True)
print(router.hit_rate())hit_rate() คือตัวเลขเดียวกับ "อัตราการพบของผิดต่อหนึ่งเบาะแส" ที่เล่าไว้ในตอนที่ 5 แต่คราวนี้แยกราย rule ได้ด้วย rule ไหนยกมือบ่อยแต่ไม่เคยเจออะไรสักที ก็ต้องกลับไปให้คนนั่งดูว่าเกณฑ์ผิด layer พื้นที่ผิด หรือพฤติกรรมในพื้นที่เปลี่ยนไปแล้ว เห้ย อย่าให้ระบบไปปรับน้ำหนักเองเงียบๆ ล่ะ เพราะนั่นคือการเอาคนออกจากวงอีกรอบนึง
แต่จะบอกอีกว่า การคำนวนพวกนี้มีหลากหลายรูปแบมากกกกก (ก.ไก่ล้านตัว) ขึ้นอยู่กับว่าเราจะเอา parameter ไหนมาใช้ ทั้งสำเร็จรูปที่มีอยู่ตามท้องตลาด กับเอาค่าจาก Data ที่เราเก็บได้มาใช้ ขึ้นอยู่กับสถานการณ์ และแน่นอน คนต้องเป็นคนตรวจสอบก่อน
ไล่กลับทั้งเรื่องแบบสั้นๆ
ทั้งหมดนี้ยังอยู่ใต้กฎเดิมตั้งแต่ตอนที่ 1 คือ ระบบยกมือ คนตัดสิน
แต่ยกมือแล้วยังไม่จบนะ ถ้ากลับไปดู Deter / Detect / Respond ของ MBC ทั้งบทความนี้เราวนอยู่แค่ Detect ส่วน Strategy ของออสเตรเลียเขียนไว้ว่าเป้าหมายของการเฝ้าระวังคือให้ได้ early warning เพื่อ "determine the most effective placement of patrol resources" คือรู้แล้วต้องวางของให้ถูกที่ด้วย
ย้อนไปที่นิยามใน MDA ขาที่สามคือ "ถ้ามันมีพิษจริง กูมีอะไรส่งไปได้บ้าง แล้วตอนนี้ของกูอยู่ตรงไหน" กับภาพเรือตรวจการณ์ลำเดียวที่กำลังเติมน้ำมันอยู่อีกฝั่งของอ่าวตอนจอขึ้นแดงแจ๋ ตอนหน้าเราจะไปต่อกันที่ Maritime Response Awareness ว่าพอวิเคราะห์ภัยคุกคามออกมาได้แล้ว จะตอบสนองภัยที่หามาได้ยังไงให้ทันและถูกลำ
เห็นมั้ยว่าทั้งหมดนี้จะเกิดขึ้นเมื่อคนมีแนวคิดก่อน คนตระหนักรู้ว่าอะไรมันมีแนวโน้ม อะไรจะเป็นภัยคุกคาม มันจะพูดยากมากกับคนที่อยู่แถวๆ นี้ที่คิดว่าเอไอมันทำได้ทุกอย่าง โยนข้อมูลให้มัน และคำตอบทุก solved หมด แค่กระดิกตีนรอ มันไม่ใช่เลยโว้ยยยยย 🤬
ระบายพอหอมปากหอมคอ ก่อนไป ขอถามคนทำงานสายนี้หน่อย เคยเจอ rule ไหนที่ยกมือทั้งวันแต่ไม่เคยเจอของจริงสักครั้งไหม หรือเคยอยู่ในโปรเจกต์ที่อยากดูทุกอย่างจนสุดท้ายไม่ได้ดูอะไรเลย มาเล่าให้ฟังหน่อย เจอกันตอนหน้าฮะ
ตอนจบของซีรีส์ ว่าด้วยการแลกเปลี่ยนข้อมูลทางทะเลข้ามประเทศ ตั้งแต่ feed แบบ near-real-time อย่าง MSSIS ศูนย์ fusion อย่าง IFC กับ IFC-IOR เครือข่ายรายงานเหตุอย่าง ReCAAP ไปจนถึงงานจริงของคนทำระบบที่ต้อง sanitize ข้อมูลก่อนปล่อยออกไป
ไล่ตระกูล INT ที่ต้องเอามาประกอบกับ sensor ตั้งแต่ OSINT HUMINT GEOINT MASINT ยัน FININT พร้อมเครื่องมือภาคสนามอย่างแอป Gather แพลตฟอร์มแบ่งปันข้อมูลอย่าง IORIS ฟอร์มตรวจค้น ไปจนถึงการหลอมทุกอย่างออกมาเป็น heat map ความเสี่ยง
ตามหาเรือที่ดับ AIS ด้วย RF Detection ชายฝั่ง RF จากดาวเทียม ดาวเทียม SAR และ EO/IR แล้วให้ Machine Learning ชี้เป้าบนภาพ ก่อนเอา detection ที่ไม่มีชื่อกลับมาจับคู่กับ track ที่หลอมไว้ตอนที่แล้ว