"""Scenario 02 — Ship-to-Ship Rendezvous: data generator. Composes shared generators from `generators/` to produce realtime, historical and static data for the S2 demo. Story: MV SAIMAA AURORA (230999081) and MV NEVA CASCADE (273999142) converge ~38 NM south of Hanko and stop alongside for ~35 minutes without filing an STS notice. A Border Guard patrol drone (RAD-DRN-PAT-01) carrying airborne MAC sensor MAC-AIR-DRN-01 orbits the meet and captures 22 transient locally-administered (02:*) MACs appearing across both hulls. Hours later 6 of those MACs are observed at Hanko port (with AURORA arriving) and 5 at Helsinki West Harbour (with CASCADE arriving) — 2 of them in BOTH port sets. """ from __future__ import annotations import json import math import random import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[2] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from generators.common import ( # noqa: E402 ambient_mmsi, crew_by_ship, load_infrastructure, load_sensors, maybe_decimate_mac_ndjson, maybe_decimate_ndjson, sensor_lookup, write_csv, write_geojson, write_ndjson, ) from generators.ais_generator import AisTrack, ais_snapshot_geojson, emit_ais # noqa: E402 from generators.mac_generator import ( # noqa: E402 MAC_CSV_HEADER, MovingMacEmitter, StaticMacEmitter, generate_background_macs, simulate_airborne_mac, simulate_moving_mac, simulate_static_mac, ) from generators.radar_generator import RadarTrack, emit_drone_radar, emit_radar # noqa: E402 UTC = timezone.utc SCENARIO_DIR = Path(__file__).resolve().parent OUT_REALTIME = SCENARIO_DIR / "data" / "realtime" OUT_STATIC = SCENARIO_DIR / "data" / "static" OUT_HISTORICAL = SCENARIO_DIR / "data" / "historical" # --------------------------------------------------------------------------- # Anchor times (per harmonized spec) # --------------------------------------------------------------------------- WINDOW_OPEN = datetime(2025, 4, 12, 8, 0, 0, tzinfo=UTC) # T-03:00 RV_START = datetime(2025, 4, 12, 11, 0, 0, tzinfo=UTC) # T+00:00 RV_END = datetime(2025, 4, 12, 11, 35, 0, tzinfo=UTC) # T+00:35 DRONE_LAUNCH = datetime(2025, 4, 12, 9, 40, 0, tzinfo=UTC) DRONE_ORBIT_START = datetime(2025, 4, 12, 10, 15, 0, tzinfo=UTC) DRONE_DESCENT_START = datetime(2025, 4, 12, 11, 8, 0, tzinfo=UTC) DRONE_DESCENT_END = datetime(2025, 4, 12, 11, 25, 0, tzinfo=UTC) DRONE_ORBIT_END = datetime(2025, 4, 12, 12, 10, 0, tzinfo=UTC) DRONE_RTB = datetime(2025, 4, 12, 12, 50, 0, tzinfo=UTC) PLANE_START = datetime(2025, 4, 12, 10, 40, 0, tzinfo=UTC) PLANE_END = datetime(2025, 4, 12, 11, 25, 0, tzinfo=UTC) HKO_ARRIVAL = datetime(2025, 4, 12, 20, 15, 0, tzinfo=UTC) HEL_ARRIVAL = datetime(2025, 4, 13, 0, 40, 0, tzinfo=UTC) WINDOW_CLOSE = datetime(2025, 4, 13, 1, 30, 0, tzinfo=UTC) RV_LAT = 59.4200 RV_LON = 23.0500 HANKO_PORT_LAT = 59.8230 HANKO_PORT_LON = 22.9700 HEL_WH_LAT = 60.1530 HEL_WH_LON = 24.9180 # --------------------------------------------------------------------------- # Vessel waypoints (per spec geometry) # --------------------------------------------------------------------------- AURORA_WAYPOINTS = [ (WINDOW_OPEN, 60.4500, 26.9500), # A1 off Kotka (datetime(2025, 4, 12, 8, 45, 0, tzinfo=UTC), 60.3500, 26.3500), # course change (datetime(2025, 4, 12, 9, 30, 0, tzinfo=UTC), 60.1800, 25.8000), # A2 mid-gulf (datetime(2025, 4, 12, 10, 15, 0, tzinfo=UTC), 59.9200, 24.9000), # transit (datetime(2025, 4, 12, 10, 30, 0, tzinfo=UTC), 59.8200, 24.6000), # A3 course-dev (datetime(2025, 4, 12, 10, 45, 0, tzinfo=UTC), 59.5500, 23.4500), # closing (RV_START, 59.4200, 23.0500), # A4 RV (RV_END, 59.4205, 23.0505), # RV drift (datetime(2025, 4, 12, 12, 0, 0, tzinfo=UTC), 59.5500, 22.9500), # post-RV (datetime(2025, 4, 12, 13, 30, 0, tzinfo=UTC), 59.6500, 22.9000), # A5 northward (datetime(2025, 4, 12, 18, 0, 0, tzinfo=UTC), 59.7800, 22.9500), # Hanko approach (HKO_ARRIVAL, 59.8230, 22.9700), # A6 berth (WINDOW_CLOSE, 59.8230, 22.9700), # stay moored ] CASCADE_WAYPOINTS = [ (WINDOW_OPEN, 59.4800, 24.8000), # B1 N of Tallinn TSS (datetime(2025, 4, 12, 9, 10, 0, tzinfo=UTC), 59.4650, 24.3000), # slowdown #1 (datetime(2025, 4, 12, 9, 30, 0, tzinfo=UTC), 59.4500, 23.9000), # B2 (datetime(2025, 4, 12, 10, 15, 0, tzinfo=UTC), 59.4300, 23.3500), # B3 slowdown (datetime(2025, 4, 12, 10, 45, 0, tzinfo=UTC), 59.4250, 23.1200), # closing (RV_START, 59.4198, 23.0506), # B4 RV (~60 m from AURORA) (RV_END, 59.4200, 23.0510), # RV drift (datetime(2025, 4, 12, 12, 0, 0, tzinfo=UTC), 59.3500, 23.4000), # post-RV (datetime(2025, 4, 12, 13, 30, 0, tzinfo=UTC), 59.3000, 23.6000), # B5 (datetime(2025, 4, 12, 18, 0, 0, tzinfo=UTC), 59.7000, 24.4000), # bend NE (datetime(2025, 4, 12, 22, 0, 0, tzinfo=UTC), 59.9500, 24.7500), # Helsinki approach (HEL_ARRIVAL, 60.1530, 24.9180), # B6 berth (WINDOW_CLOSE, 60.1530, 24.9180), # stay moored ] # --------------------------------------------------------------------------- # Drone orbit (RAD-DRN-PAT-01) # --------------------------------------------------------------------------- DRONE_BASE_LAT = 59.8240 # Hanko Coast Guard Station (synthetic) DRONE_BASE_LON = 22.9650 DRONE_ORBIT_CENTER_LAT = 59.4205 DRONE_ORBIT_CENTER_LON = 23.0510 def build_drone_track() -> list[tuple[datetime, float, float]]: """Drone: base → en-route → 1.5 km orbit → en-route → base.""" wps: list[tuple[datetime, float, float]] = [] wps.append((DRONE_LAUNCH, DRONE_BASE_LAT, DRONE_BASE_LON)) # En-route (straight-line) — 35 min from base to orbit centre wps.append((DRONE_ORBIT_START, DRONE_ORBIT_CENTER_LAT, DRONE_ORBIT_CENTER_LON)) # Orbit phase — generate circular waypoints radius_m = 1500.0 period_s = 336.0 # ~5.6 min per orbit at ~28 m/s on a 9425 m circumference step_s = 10.0 t = DRONE_ORBIT_START while t <= DRONE_ORBIT_END: elapsed = (t - DRONE_ORBIT_START).total_seconds() angle = 2 * math.pi * (elapsed / period_s) # Approx local conversion (m → deg) dlat = (radius_m * math.cos(angle)) / 111_111.0 dlon = (radius_m * math.sin(angle)) / ( 111_111.0 * math.cos(math.radians(DRONE_ORBIT_CENTER_LAT)) ) wps.append((t, DRONE_ORBIT_CENTER_LAT + dlat, DRONE_ORBIT_CENTER_LON + dlon)) t = t + timedelta(seconds=step_s) # RTB (straight-line) — 40 min back to base wps.append((DRONE_RTB, DRONE_BASE_LAT, DRONE_BASE_LON)) return wps def drone_altitudes() -> list[tuple[datetime, float]]: return [ (DRONE_LAUNCH, 50.0), (DRONE_ORBIT_START, 1200.0), (DRONE_DESCENT_START, 1200.0), (datetime(2025, 4, 12, 11, 12, 0, tzinfo=UTC), 700.0), (DRONE_DESCENT_END, 1200.0), (DRONE_ORBIT_END, 1200.0), (DRONE_RTB, 30.0), ] # --------------------------------------------------------------------------- # Plane (RAD-PLN-01) — Dornier 228-class peripheral transit # --------------------------------------------------------------------------- PLANE_WAYPOINTS = [ (PLANE_START, 59.1500, 22.6000), (datetime(2025, 4, 12, 11, 5, 0, tzinfo=UTC), 59.1000, 23.0000), (PLANE_END, 59.0800, 23.6000), ] # --------------------------------------------------------------------------- # Burst MACs (locally-administered 02:*, manufacturer = null) # --------------------------------------------------------------------------- BURST_RNG = random.Random(2025_04_12_2) # deterministic re-runs def make_burst_macs(n: int) -> list[str]: macs: list[str] = [] seen: set[str] = set() while len(macs) < n: mac = "02:" + ":".join(f"{BURST_RNG.randint(0, 255):02x}" for _ in range(5)) if mac in seen: continue seen.add(mac) macs.append(mac) return macs BURST_MACS = make_burst_macs(22) # Indices: 0..21 # 6 burst MACs at Hanko port: BURST_MACS[0:6] # 5 burst MACs at Helsinki, with 2 overlapping the Hanko set (indices 4,5): # BURST_MACS[4:9] HANKO_BURST_SUBSET = BURST_MACS[0:6] HEL_BURST_SUBSET = BURST_MACS[4:9] # --------------------------------------------------------------------------- # Extended persistent crew MACs per ship (catalog block + synthetic extensions # using catalog OUI prefixes; suffix scheme per spec — `..:23..29` etc.) # --------------------------------------------------------------------------- AURORA_PERSONAS_EXTRA = [ {"id": "P-SAU-BOSUN", "device_mac": "B0:7D:64:A1:5B:23", "oui_vendor": "Apple"}, {"id": "P-SAU-ENG", "device_mac": "8C:16:45:33:44:24", "oui_vendor": "Lenovo"}, {"id": "P-SAU-COOK", "device_mac": "04:CF:8C:55:66:25", "oui_vendor": "Xiaomi"}, {"id": "P-SAU-AB1", "device_mac": "38:F9:D3:11:22:26", "oui_vendor": "Samsung"}, {"id": "P-SAU-AB2", "device_mac": "00:E0:FC:77:88:27", "oui_vendor": "Huawei"}, {"id": "P-SAU-CADET", "device_mac": "A4:83:E7:5C:9B:28", "oui_vendor": "Apple"}, {"id": "P-SAU-OILER", "device_mac": "8C:16:45:33:44:29", "oui_vendor": "Lenovo"}, ] CASCADE_PERSONAS_EXTRA = [ {"id": "P-NEV-BOSUN", "device_mac": "B0:7D:64:A1:5B:33", "oui_vendor": "Apple"}, {"id": "P-NEV-ENG", "device_mac": "8C:16:45:33:44:34", "oui_vendor": "Lenovo"}, {"id": "P-NEV-COOK", "device_mac": "04:CF:8C:55:66:35", "oui_vendor": "Xiaomi"}, {"id": "P-NEV-AB1", "device_mac": "38:F9:D3:11:22:36", "oui_vendor": "Samsung"}, {"id": "P-NEV-AB2", "device_mac": "00:E0:FC:77:88:37", "oui_vendor": "Huawei"}, {"id": "P-NEV-CADET", "device_mac": "A4:83:E7:5C:9B:38", "oui_vendor": "Apple"}, {"id": "P-NEV-OILER", "device_mac": "8C:16:45:33:44:39", "oui_vendor": "Lenovo"}, ] def aurora_crew() -> list[dict[str, Any]]: base = [p for p in crew_by_ship("MV SAIMAA AURORA")] return base + AURORA_PERSONAS_EXTRA def cascade_crew() -> list[dict[str, Any]]: base = [p for p in crew_by_ship("MV NEVA CASCADE")] return base + CASCADE_PERSONAS_EXTRA # --------------------------------------------------------------------------- # Sensors / infrastructure subsets used in this scenario # --------------------------------------------------------------------------- SENSORS_USED_IDS = { "MAC-AIR-DRN-01", "MAC-AIR-PLN-01", "MAC-HKO-PORT-01", "MAC-HKO-PORT-02", "MAC-HKO-PORT-03", "MAC-HEL-PORT-01", "MAC-HEL-PORT-02", "MAC-HKO-COAST-01", "MAC-UTO-COAST-01", "MAC-PRK-COAST-01", "MAC-PRK-COAST-02", "MAC-INK-COAST-01", "RAD-COAST-HKO-01", "RAD-COAST-HEL-01", "RAD-PLN-01", "RAD-DRN-PAT-01", } INFRA_USED_IDS = { "finnish-eez-gof", "shipping-lane-eb", "shipping-lane-wb", "port-hanko", "port-helsinki", } # --------------------------------------------------------------------------- # Ambient AIS to fill realtime volume + decoy STS pair # --------------------------------------------------------------------------- def build_ambient_ais(n_ships: int, seed: int) -> list[dict[str, Any]]: """Random cross-gulf ambient transits during the RV window. Cadence 15 s. Uses the synthetic 9XX ambient MMSI block from `personas.json` so they cannot collide with real ITU MIDs or with catalog protagonists. """ rng = random.Random(seed) out: list[dict[str, Any]] = [] for i in range(n_ships): eastbound = rng.random() < 0.5 lat0 = rng.uniform(59.55, 60.25) lat1 = lat0 + rng.uniform(-0.08, 0.08) if eastbound: lon0, lon1 = 22.5, 27.5 else: lon0, lon1 = 27.5, 22.5 t_start = WINDOW_OPEN + timedelta(minutes=rng.uniform(0, 240)) t_end = min(WINDOW_CLOSE, t_start + timedelta(minutes=rng.uniform(120, 300))) if t_end <= t_start: continue flag_roll = rng.random() flag = "FI" if flag_roll < 0.65 else ("EE" if flag_roll < 0.85 else "OTHER") mmsi = ambient_mmsi(rng, flag) track = AisTrack( mmsi=mmsi, waypoints=[(t_start, lat0, lon0), (t_end, lat1, lon1)], cadence_s=15.0, destination="FIHEL" if eastbound else "EETLL", seed=seed + i, ) out.extend(emit_ais(track)) return out # Decoy STS pair (legitimate OPA-90 lightering ~30 NM ENE of RV). # These vessels are scenario-local — not added to personas.json — and only # appear in the historical baseline for 2025-04-05. DECOY_NORDLYS_MMSI = 230888301 DECOY_BALTIC_MMSI = 256888422 DECOY_LAT = 59.6500 DECOY_LON = 23.7500 # --------------------------------------------------------------------------- # Realtime generation # --------------------------------------------------------------------------- def _ais_for(mmsi: int, name: str, callsign: str, imo: int, ship_type: int, loa: int, beam: int, waypoints, dest: str, seed: int, cadence_s: float) -> list[dict[str, Any]]: """Emit AIS records for a vessel not in personas catalog (decoy).""" rng = random.Random(seed) out: list[dict[str, Any]] = [] from generators.common import time_range, interp_waypoints, iso_utc start = waypoints[0][0] end = waypoints[-1][0] for t in time_range(start, end, cadence_s): lat, lon, sog_kn, cog = interp_waypoints(waypoints, t) sog_kn = max(0.0, sog_kn + rng.gauss(0.0, 0.2)) cog = (cog + rng.gauss(0.0, 1.5)) % 360.0 out.append({ "timestamp": iso_utc(t), "ts_epoch_ms": int(t.timestamp() * 1000), "mmsi": mmsi, "imo": imo, "name": name, "callsign": callsign, "ship_type": ship_type, "loa_m": loa, "beam_m": beam, "lat": round(lat, 6), "lon": round(lon, 6), "sog_kn": round(sog_kn, 2), "cog_deg": round(cog, 1), "heading_deg": round(cog, 0), "nav_status": 0, "destination": dest, "msg_class": "A", "source_receiver": "AIS-COAST-FIN", }) return out def generate_realtime() -> dict[str, int]: """Generate the realtime layer. Returns stream → row counts (incl. disclaimer).""" sensors = sensor_lookup() counts: dict[str, int] = {} # ----- AIS (both ships + ambient) ----- aurora_track = AisTrack( mmsi=230999081, waypoints=AURORA_WAYPOINTS, cadence_s=10.0, destination="FIHAN", nav_status=0, seed=11001, ) aurora_msgs = emit_ais(aurora_track) cascade_track = AisTrack( mmsi=273999142, waypoints=CASCADE_WAYPOINTS, cadence_s=10.0, destination="FIHEL", nav_status=0, seed=11002, ) cascade_msgs = emit_ais(cascade_track) ambient_msgs = build_ambient_ais(n_ships=1500, seed=11003) ais_all = aurora_msgs + cascade_msgs + ambient_msgs counts["ais.ndjson"] = write_ndjson( OUT_REALTIME / "ais.ndjson", ais_all, "s2-rendezvous/ais") snapshot_features = ais_snapshot_geojson(ais_all) counts["ais_snapshot.geojson"] = write_geojson( OUT_REALTIME / "ais_snapshot.geojson", snapshot_features, "s2-rendezvous/ais_snapshot") # ----- Drone radar (RAD-DRN-PAT-01) ----- drone_track_wps = build_drone_track() drone_radar = RadarTrack( track_id="DRN-PAT-01", sensor_id="RAD-DRN-PAT-01", waypoints=drone_track_wps, cadence_s=2.0, classification="airframe_uav", rcs_m2=0.05, confidence=0.95, seed=12001, ) drone_radar_msgs = emit_drone_radar(drone_radar, altitudes=drone_altitudes()) for r in drone_radar_msgs: r["platform"] = "RAD-DRN-PAT-01" counts["drone_radar.ndjson"] = write_ndjson( OUT_REALTIME / "drone_radar.ndjson", drone_radar_msgs, "s2-rendezvous/drone_radar") # ----- Drone radar TARGET plots (AURORA + CASCADE seen from drone) ----- # Reuse the AIS waypoints as ground-truth target tracks, emit at 2 s cadence # during the orbit window. drone_target_window = [ (DRONE_ORBIT_START, AURORA_WAYPOINTS), (DRONE_ORBIT_START, CASCADE_WAYPOINTS), ] drone_target_msgs: list[dict[str, Any]] = [] for tgt_id, tgt_mmsi, wps in [ ("TGT-A", 230999081, AURORA_WAYPOINTS), ("TGT-B", 273999142, CASCADE_WAYPOINTS), ]: # Clip waypoints to drone orbit window clipped = [(t, lat, lon) for (t, lat, lon) in wps if DRONE_ORBIT_START <= t <= DRONE_ORBIT_END] # Make sure we have an interpolated start/end inside the window from generators.common import interp_waypoints if not clipped or clipped[0][0] > DRONE_ORBIT_START: lat, lon, _, _ = interp_waypoints(wps, DRONE_ORBIT_START) clipped.insert(0, (DRONE_ORBIT_START, lat, lon)) if clipped[-1][0] < DRONE_ORBIT_END: lat, lon, _, _ = interp_waypoints(wps, DRONE_ORBIT_END) clipped.append((DRONE_ORBIT_END, lat, lon)) tgt = RadarTrack( track_id=tgt_id, sensor_id="RAD-DRN-PAT-01", waypoints=clipped, cadence_s=2.0, classification="surface_large", rcs_m2=1800.0, confidence=0.88, seed=12100 + (1 if tgt_id == "TGT-A" else 2), ) tmsgs = emit_radar(tgt) for r in tmsgs: r["mmsi_hint"] = tgt_mmsi r["platform"] = "RAD-DRN-PAT-01" drone_target_msgs.extend(tmsgs) counts["drone_radar_targets.ndjson"] = write_ndjson( OUT_REALTIME / "drone_radar_targets.ndjson", drone_target_msgs, "s2-rendezvous/drone_radar_targets") # ----- Plane radar (RAD-PLN-01) peripheral ----- plane_radar = RadarTrack( track_id="PLN-TRK-S2-0044", sensor_id="RAD-PLN-01", waypoints=PLANE_WAYPOINTS, cadence_s=4.0, classification="surface_large", rcs_m2=6500.0, confidence=0.78, seed=13001, ) plane_msgs = emit_radar(plane_radar) for r in plane_msgs: r["platform"] = "RAD-PLN-01" counts["plane_radar.ndjson"] = write_ndjson( OUT_REALTIME / "plane_radar.ndjson", plane_msgs, "s2-rendezvous/plane_radar") # ----- MAC observations ----- mac_obs = [] # 1) Persistent crew MACs riding each ship (visible to coastal + port sensors) for i, p in enumerate(aurora_crew()): em = MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=AURORA_WAYPOINTS, active_windows=[(WINDOW_OPEN, WINDOW_CLOSE)], seed=20000 + i, ) mac_obs.extend(simulate_moving_mac(em, sensors, session_window_s=180.0)) for i, p in enumerate(cascade_crew()): em = MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=CASCADE_WAYPOINTS, active_windows=[(WINDOW_OPEN, WINDOW_CLOSE)], seed=21000 + i, ) mac_obs.extend(simulate_moving_mac(em, sensors, session_window_s=180.0)) # 2) Airborne MAC sensor MAC-AIR-DRN-01 — primary capture at RV # Target list: 4 most-prominent persistent crew (2 per ship) + # 22 burst MACs. Drone track is the orbit; we pass two slightly # different drifting points for the bursts so RSSI varies per ship. burst_track_a = [ # bursts visible at/near AURORA RV position (RV_START - timedelta(minutes=15), 59.4205, 23.0512), (RV_START, 59.4205, 23.0510), (RV_END, 59.4208, 23.0515), (RV_END + timedelta(minutes=5), 59.4210, 23.0520), ] burst_track_b = [ # bursts visible at/near CASCADE RV position (RV_START - timedelta(minutes=15), 59.4196, 23.0500), (RV_START, 59.4198, 23.0506), (RV_END, 59.4201, 23.0510), (RV_END + timedelta(minutes=5), 59.4203, 23.0512), ] burst_window_start = RV_START - timedelta(minutes=5) burst_window_end = RV_END + timedelta(minutes=5) burst_emitters: list[MovingMacEmitter] = [] for idx, mac in enumerate(BURST_MACS): wps = burst_track_a if (idx % 2 == 0) else burst_track_b burst_emitters.append(MovingMacEmitter( mac=mac, manufacturer=None, # null per spec waypoints=wps, active_windows=[(burst_window_start, burst_window_end)], seed=30000 + idx, )) # Crew priors visible from drone (top 4 crew = master + chief of each ship) crew_priors_aurora = aurora_crew()[:2] crew_priors_cascade = cascade_crew()[:2] crew_drone_emitters = [ MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=AURORA_WAYPOINTS, active_windows=[(DRONE_ORBIT_START, DRONE_ORBIT_END)], seed=40000 + i, ) for i, p in enumerate(crew_priors_aurora) ] + [ MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=CASCADE_WAYPOINTS, active_windows=[(DRONE_ORBIT_START, DRONE_ORBIT_END)], seed=41000 + i, ) for i, p in enumerate(crew_priors_cascade) ] # Clip drone track to the orbit window so we only emit airborne sessions there orbit_drone_track = [(t, lat, lon) for (t, lat, lon) in drone_track_wps if DRONE_ORBIT_START <= t <= DRONE_ORBIT_END] # Airborne MAC observations — burst (dense, 1 s windows during RV ±5 min) burst_drone_track = [(t, lat, lon) for (t, lat, lon) in orbit_drone_track if burst_window_start <= t <= burst_window_end] if not burst_drone_track: burst_drone_track = orbit_drone_track[:1] mac_obs.extend(simulate_airborne_mac( "MAC-AIR-DRN-01", burst_drone_track, burst_emitters, session_window_s=2.0, rssi_threshold=-95.0, )) # Crew prior observations from drone — coarser (5 s windows) over full orbit mac_obs.extend(simulate_airborne_mac( "MAC-AIR-DRN-01", orbit_drone_track, crew_drone_emitters, session_window_s=5.0, rssi_threshold=-95.0, )) # 3) Airborne MAC sensor MAC-AIR-PLN-01 — peripheral, path-loss limited plane_mac_emitters = [ MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=AURORA_WAYPOINTS, active_windows=[(PLANE_START, PLANE_END)], seed=42000 + i, ) for i, p in enumerate(crew_priors_aurora) ] + [ MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=CASCADE_WAYPOINTS, active_windows=[(PLANE_START, PLANE_END)], seed=43000 + i, ) for i, p in enumerate(crew_priors_cascade) ] plane_mac_track = [(t, lat, lon) for (t, lat, lon) in PLANE_WAYPOINTS] mac_obs.extend(simulate_airborne_mac( "MAC-AIR-PLN-01", plane_mac_track, plane_mac_emitters, session_window_s=10.0, rssi_threshold=-100.0, )) # 4) Cross-port re-appearance — bursts seen at port sensors hours later hanko_sensors = { sid: sensors[sid] for sid in ("MAC-HKO-PORT-01", "MAC-HKO-PORT-02", "MAC-HKO-PORT-03") } hel_sensors = { sid: sensors[sid] for sid in ("MAC-HEL-PORT-01", "MAC-HEL-PORT-02") } # Hanko: ±90 min window around AURORA berth hko_active = [(HKO_ARRIVAL - timedelta(minutes=30), HKO_ARRIVAL + timedelta(minutes=90))] for idx, mac in enumerate(HANKO_BURST_SUBSET): em = StaticMacEmitter( mac=mac, manufacturer=None, lat=HANKO_PORT_LAT, lon=HANKO_PORT_LON, active_windows=hko_active, seed=50000 + idx, ) mac_obs.extend(simulate_static_mac(em, hanko_sensors, session_window_s=120.0)) hel_active = [(HEL_ARRIVAL - timedelta(minutes=30), HEL_ARRIVAL + timedelta(minutes=90))] for idx, mac in enumerate(HEL_BURST_SUBSET): em = StaticMacEmitter( mac=mac, manufacturer=None, lat=HEL_WH_LAT, lon=HEL_WH_LON, active_windows=hel_active, seed=51000 + idx, ) mac_obs.extend(simulate_static_mac(em, hel_sensors, session_window_s=120.0)) # 5) Background MAC noise across coastal/port sensors for the whole window. bg = generate_background_macs( sensors, WINDOW_OPEN, WINDOW_CLOSE, mac_count=180, cadence_s=90.0, seed=60000, ) mac_obs.extend(bg) mac_nd = [m.to_ndjson() for m in mac_obs] counts["mac.ndjson"] = write_ndjson( OUT_REALTIME / "mac.ndjson", mac_nd, "s2-rendezvous/mac") mac_rows = [m.to_csv_row() for m in mac_obs] counts["mac.csv"] = write_csv( OUT_REALTIME / "mac.csv", MAC_CSV_HEADER, mac_rows, "s2-rendezvous/mac_sessions") # Decimated companion files for any realtime NDJSON > 20 MB decim_reports = [] AIS_DECIM_FIELDS = ["timestamp", "lat", "lon", "sog_kn", "cog_deg", "nav_status"] RADAR_DECIM_FIELDS = ["timestamp", "lat", "lon", "sog_kn", "cog_deg", "alt_m", "speed_mps", "heading_deg", "rcs_m2", "classification", "mmsi_hint", "kind"] for path, kw in [ (OUT_REALTIME / "ais.ndjson", {"key_field": "mmsi", "ts_field": "ts_epoch_ms", "project_fields": AIS_DECIM_FIELDS}), (OUT_REALTIME / "plane_radar.ndjson", {"key_field": "track_id", "ts_field": "ts_epoch_ms", "project_fields": RADAR_DECIM_FIELDS}), (OUT_REALTIME / "drone_radar.ndjson", {"key_field": "track_id", "ts_field": "ts_epoch_ms", "project_fields": RADAR_DECIM_FIELDS}), (OUT_REALTIME / "drone_radar_targets.ndjson", {"key_field": "track_id", "ts_field": "ts_epoch_ms", "project_fields": RADAR_DECIM_FIELDS}), ]: rep = maybe_decimate_ndjson(path, **kw) if rep: decim_reports.append(rep) counts[Path(rep["decimated"]).name] = rep["rows"] + 1 mac_rep = maybe_decimate_mac_ndjson(OUT_REALTIME / "mac.ndjson") if mac_rep: decim_reports.append(mac_rep) counts[Path(mac_rep["decimated"]).name] = mac_rep["rows"] + 1 if decim_reports: print("[S2] decimated companion files:") for r in decim_reports: print(f" {Path(r['decimated']).name} " f"{r['source_bytes']/1024/1024:.1f}MB -> {r['decimated_bytes']/1024/1024:.1f}MB" f" ({r['rows']} rows)") return counts # --------------------------------------------------------------------------- # Static layer # --------------------------------------------------------------------------- def generate_static() -> dict[str, int]: counts: dict[str, int] = {} # area_of_interest — Gulf of Finland bbox covering RV + both arrival ports aoi = { "type": "Feature", "properties": { "featureId": "s2-aoi", "name": "S2 Area of Interest", "note": "Bounding polygon covering RV point, both ship transit lanes, and Hanko + Helsinki West Harbour port clusters", }, "geometry": { "type": "Polygon", "coordinates": [[ [22.5, 59.30], [25.5, 59.30], [25.5, 60.35], [22.5, 60.35], [22.5, 59.30], ]], }, } counts["area_of_interest.geojson"] = write_geojson( OUT_STATIC / "area_of_interest.geojson", [aoi], "s2-rendezvous/area_of_interest") # sensors_used — subset of sensors catalog sensors_fc = load_sensors() sensor_feats = [f for f in sensors_fc["features"] if f["properties"]["sensorId"] in SENSORS_USED_IDS] counts["sensors_used.geojson"] = write_geojson( OUT_STATIC / "sensors_used.geojson", sensor_feats, "s2-rendezvous/sensors_used") # infrastructure_used — subset of infrastructure catalog infra_fc = load_infrastructure() infra_feats = [f for f in infra_fc["features"] if f["properties"]["featureId"] in INFRA_USED_IDS] counts["infrastructure_used.geojson"] = write_geojson( OUT_STATIC / "infrastructure_used.geojson", infra_feats, "s2-rendezvous/infrastructure_used") # rendezvous_zone — three concentric circles around the RV point def circle_polygon(lat: float, lon: float, radius_m: float, n: int = 64): coords = [] for i in range(n): angle = 2 * math.pi * (i / n) dlat = (radius_m * math.cos(angle)) / 111_111.0 dlon = (radius_m * math.sin(angle)) / (111_111.0 * math.cos(math.radians(lat))) coords.append([lon + dlon, lat + dlat]) coords.append(coords[0]) return coords rv_zones = [ {"type": "Feature", "properties": {"featureId": f"s2-rv-{label}", "name": label, "radius_m": r}, "geometry": {"type": "Polygon", "coordinates": [circle_polygon(RV_LAT, RV_LON, r)]}} for label, r in [("alert-100m", 100.0), ("watch-500m", 500.0), ("context-2NM", 3704.0)] ] counts["rendezvous_zone.geojson"] = write_geojson( OUT_STATIC / "rendezvous_zone.geojson", rv_zones, "s2-rendezvous/rendezvous_zone") # ship_a / ship_b routes — LineString features for the scenario tracks ship_a_feat = { "type": "Feature", "properties": {"featureId": "ship-a-rv-route", "mmsi": 230999081, "name": "MV SAIMAA AURORA scenario track"}, "geometry": {"type": "LineString", "coordinates": [[lon, lat] for (_, lat, lon) in AURORA_WAYPOINTS]}, } ship_b_feat = { "type": "Feature", "properties": {"featureId": "ship-b-rv-route", "mmsi": 273999142, "name": "MV NEVA CASCADE scenario track"}, "geometry": {"type": "LineString", "coordinates": [[lon, lat] for (_, lat, lon) in CASCADE_WAYPOINTS]}, } counts["ship_routes.geojson"] = write_geojson( OUT_STATIC / "ship_routes.geojson", [ship_a_feat, ship_b_feat], "s2-rendezvous/ship_routes") # drone_orbit_path — LineString of the orbit waypoints (centre + radius) drone_path = build_drone_track() drone_feat = { "type": "Feature", "properties": {"featureId": "drone-orbit-path", "sensorId": "RAD-DRN-PAT-01", "carries_sensor": "MAC-AIR-DRN-01", "orbit_centre": [DRONE_ORBIT_CENTER_LON, DRONE_ORBIT_CENTER_LAT], "orbit_radius_m": 1500}, "geometry": {"type": "LineString", "coordinates": [[lon, lat] for (_, lat, lon) in drone_path]}, } plane_feat = { "type": "Feature", "properties": {"featureId": "plane-transit-path", "sensorId": "RAD-PLN-01", "carries_sensor": "MAC-AIR-PLN-01"}, "geometry": {"type": "LineString", "coordinates": [[lon, lat] for (_, lat, lon) in PLANE_WAYPOINTS]}, } counts["airborne_platforms.geojson"] = write_geojson( OUT_STATIC / "airborne_platforms.geojson", [drone_feat, plane_feat], "s2-rendezvous/airborne_platforms") # Decoy STS lightering geometry (baseline-day legitimate event) decoy_feat = { "type": "Feature", "properties": { "featureId": "decoy-sts-lightering", "name": "Legitimate OPA-90 STS lightering (decoy)", "date": "2025-04-05", "vessels": [DECOY_NORDLYS_MMSI, DECOY_BALTIC_MMSI], "note": "Filed STS notice ≥96 h ahead; not overflown by RAD-DRN-PAT-01; no burst MACs.", }, "geometry": {"type": "Point", "coordinates": [DECOY_LON, DECOY_LAT]}, } counts["decoy_sts_lightering.geojson"] = write_geojson( OUT_STATIC / "decoy_sts_lightering.geojson", [decoy_feat], "s2-rendezvous/decoy_sts_lightering") return counts # --------------------------------------------------------------------------- # Historical baseline (3 weeks before RV) + decoy STS # --------------------------------------------------------------------------- def _shift(wps: list[tuple[datetime, float, float]], dt: timedelta) -> list[tuple[datetime, float, float]]: return [(t + dt, lat, lon) for (t, lat, lon) in wps] def generate_historical() -> dict[str, int]: """Per-day baseline transits for both ships + decoy STS on 2025-04-05. To keep the script reasonable the baseline replays an abridged daily transit for each ship for 14 days; crew MACs co-occur with AIS at the relevant port sensors. No burst MACs in baseline (proves novelty). """ counts: dict[str, int] = {} sensors = sensor_lookup() ais_baseline: list[dict[str, Any]] = [] mac_baseline_obs = [] # Use a simplified daily route for AURORA (Kotka -> Hanko) and CASCADE # (Tallinn -> Helsinki) covering 6 hours per day, 30 s cadence. def abridged_aurora(day_anchor: datetime) -> list[tuple[datetime, float, float]]: return [ (day_anchor, 60.4500, 26.9500), (day_anchor + timedelta(hours=2), 60.0500, 25.5000), (day_anchor + timedelta(hours=4), 59.9000, 24.0000), (day_anchor + timedelta(hours=5, minutes=30), 59.8230, 22.9700), ] def abridged_cascade(day_anchor: datetime) -> list[tuple[datetime, float, float]]: return [ (day_anchor, 59.4800, 24.8000), (day_anchor + timedelta(hours=2), 59.6500, 24.7000), (day_anchor + timedelta(hours=4), 59.9500, 24.7500), (day_anchor + timedelta(hours=5, minutes=30), 60.1530, 24.9180), ] baseline_days = [WINDOW_OPEN.replace(hour=6, minute=0, second=0, microsecond=0) - timedelta(days=d) for d in range(1, 15)] for d in baseline_days: aur_wp = abridged_aurora(d) cas_wp = abridged_cascade(d) aur_hist = AisTrack( mmsi=230999081, waypoints=aur_wp, cadence_s=30.0, destination="FIHAN", seed=70000 + d.day, ) ais_baseline.extend(emit_ais(aur_hist)) cas_hist = AisTrack( mmsi=273999142, waypoints=cas_wp, cadence_s=30.0, destination="FIHEL", seed=71000 + d.day, ) ais_baseline.extend(emit_ais(cas_hist)) # Crew MAC co-occurrence (just top 4 per ship to bound volume) for i, p in enumerate(aurora_crew()[:4]): em = MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=aur_wp, active_windows=[(aur_wp[0][0], aur_wp[-1][0])], seed=72000 + d.day * 10 + i, ) mac_baseline_obs.extend(simulate_moving_mac(em, sensors, session_window_s=300.0)) for i, p in enumerate(cascade_crew()[:4]): em = MovingMacEmitter( mac=p["device_mac"], manufacturer=p.get("oui_vendor"), waypoints=cas_wp, active_windows=[(cas_wp[0][0], cas_wp[-1][0])], seed=73000 + d.day * 10 + i, ) mac_baseline_obs.extend(simulate_moving_mac(em, sensors, session_window_s=300.0)) # Per-day background MAC noise (lighter than realtime) bg = generate_background_macs( sensors, d, d + timedelta(hours=6), mac_count=40, cadence_s=300.0, seed=80000 + d.day, ) mac_baseline_obs.extend(bg) # -------------------- Decoy STS lightering (2025-04-05) ------------------- decoy_day = datetime(2025, 4, 5, 9, 0, 0, tzinfo=UTC) decoy_end = datetime(2025, 4, 5, 15, 0, 0, tzinfo=UTC) # Both decoy vessels meet at DECOY_LAT/DECOY_LON, perform a clean STS # lightering with filed notice and proper nav_status. They are tankers # (type 80). Pair tracks: NORDLYS arrives, BALTIC waits alongside. nordlys_wp = [ (decoy_day - timedelta(hours=2), 59.55, 24.40), (decoy_day, DECOY_LAT, DECOY_LON), (decoy_end, DECOY_LAT, DECOY_LON), (decoy_end + timedelta(hours=2), 59.50, 23.20), ] baltic_wp = [ (decoy_day - timedelta(hours=2), 59.45, 22.80), (decoy_day - timedelta(minutes=30), DECOY_LAT, DECOY_LON), (decoy_end, DECOY_LAT, DECOY_LON), (decoy_end + timedelta(hours=2), 59.70, 24.50), ] decoy_msgs = [] decoy_msgs.extend(_ais_for( DECOY_NORDLYS_MMSI, "MV NORDLYS HARMONY", "OJZZ7", 8888301, 80, 148, 22, nordlys_wp, "STS-OP", seed=90001, cadence_s=30.0, )) decoy_msgs.extend(_ais_for( DECOY_BALTIC_MMSI, "MV BALTIC EMERALD", "9HZZ8", 8888422, 80, 152, 24, baltic_wp, "STS-OP", seed=90002, cadence_s=30.0, )) # Mark NORDLYS as 'moored' during the STS window (nav_status 5) for r in decoy_msgs: if r["mmsi"] == DECOY_NORDLYS_MMSI: ts = datetime.strptime(r["timestamp"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=UTC) if decoy_day <= ts < decoy_end: r["nav_status"] = 5 # moored # Decoy MAC: each tanker has 2 persistent crew MACs visible to coastal # sensors during transit, but ZERO burst MACs. DECOY_CREW = [ ("a4:83:e7:5c:9b:91", "Apple", nordlys_wp), ("38:f9:d3:11:22:91", "Samsung", nordlys_wp), ("00:e0:fc:77:88:92", "Huawei", baltic_wp), ("04:cf:8c:55:66:92", "Xiaomi", baltic_wp), ] for i, (mac, vendor, wp) in enumerate(DECOY_CREW): em = MovingMacEmitter( mac=mac, manufacturer=vendor, waypoints=wp, active_windows=[(wp[0][0], wp[-1][0])], seed=91000 + i, ) mac_baseline_obs.extend(simulate_moving_mac(em, sensors, session_window_s=300.0)) ais_baseline.extend(decoy_msgs) counts["ais_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "ais_baseline.ndjson", ais_baseline, "s2-rendezvous/ais_baseline") mac_nd = [m.to_ndjson() for m in mac_baseline_obs] counts["mac_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "mac_baseline.ndjson", mac_nd, "s2-rendezvous/mac_baseline") mac_rows = [m.to_csv_row() for m in mac_baseline_obs] counts["mac_baseline.csv"] = write_csv( OUT_HISTORICAL / "mac_baseline.csv", MAC_CSV_HEADER, mac_rows, "s2-rendezvous/mac_baseline_sessions") return counts def dir_size_bytes(p: Path) -> int: return sum(f.stat().st_size for f in p.rglob("*") if f.is_file()) def main() -> int: print("[S2] burst MACs (22):") for mac in BURST_MACS: tag = [] if mac in HANKO_BURST_SUBSET: tag.append("HKO") if mac in HEL_BURST_SUBSET: tag.append("HEL") print(f" {mac} {','.join(tag) or '-'}") print("[S2] generating realtime layer …") rt = generate_realtime() print("[S2] generating static layer …") st = generate_static() print("[S2] generating historical layer …") hi = generate_historical() print("\n===== Scenario 02 — Ship-to-Ship Rendezvous: generation summary =====") print("\n[realtime]") for k, v in rt.items(): print(f" {k:<32} rows={v:>10}") print("\n[static]") for k, v in st.items(): print(f" {k:<32} features={v:>6}") print("\n[historical]") for k, v in hi.items(): print(f" {k:<32} rows={v:>10}") rt_bytes = dir_size_bytes(OUT_REALTIME) st_bytes = dir_size_bytes(OUT_STATIC) hi_bytes = dir_size_bytes(OUT_HISTORICAL) print("\n[on disk]") print(f" realtime {rt_bytes:>12} bytes ({rt_bytes/1024/1024:.2f} MB)") print(f" static {st_bytes:>12} bytes ({st_bytes/1024:.2f} KB)") print(f" historical {hi_bytes:>12} bytes ({hi_bytes/1024/1024:.2f} MB)") total_rt = sum(v for k, v in rt.items() if k.endswith((".ndjson", ".csv"))) total_hi = sum(v for k, v in hi.items() if k.endswith((".ndjson", ".csv"))) print(f"\n[totals] realtime rows={total_rt} historical rows={total_hi}") print("[done] All files written under scenarios/02-ship-to-ship-rendezvous/data/") summary = { "scenario": "s2-ship-to-ship-rendezvous", "burst_macs": BURST_MACS, "hanko_burst_subset": HANKO_BURST_SUBSET, "helsinki_burst_subset": HEL_BURST_SUBSET, "cross_port_overlap": sorted(set(HANKO_BURST_SUBSET) & set(HEL_BURST_SUBSET)), "realtime": rt, "static": st, "historical": hi, "bytes": {"realtime": rt_bytes, "static": st_bytes, "historical": hi_bytes}, } (SCENARIO_DIR / "data" / "_generation_summary.json").write_text( json.dumps(summary, indent=2), encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())