"""Scenario 08 — Red Vessel Escalation: data generator. Story: MV KASPIYSK (MMSI 636092744, Liberia-flagged, IMO on watchlist) enters the Gulf of Finland from the east and transits directly toward Helsinki. The vessel crosses the Finnish EEZ boundary at 07:30 UTC, slows to 4 kn at 08:30 as coast-guard radar acquires it, then holds at a suspicious anchor position ~18 km south of Helsinki. At 09:15 MAC sensors detect the drone-controller MAC and five drone-boot MACs — the smoking-gun pre-launch observation. At 09:25 the ship goes AIS-dark; at 09:30 five DJI drones lift off in a fan toward Helsinki-area targets (Espoo coast, Lauttasaari, West Harbor, Helsinki center, Kulosaari). AIS briefly reappears at 09:35 then goes dark again until 10:15 when the ship re-appears heading south at 12 kn. EEZ exit at 11:30; window closes 14:00 UTC. This is the longest scenario in the dataset (8 hours) and the only one with a Liberia-flagged vessel on the IMO watchlist triggering an initial yellow alert before any overt hostile action is observed. """ 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 KN_MS, ambient_mmsi, haversine_m, iso_utc, load_infrastructure, load_sensors, maybe_decimate_mac_ndjson, maybe_decimate_ndjson, rssi_from_distance, 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, MacObservation, MovingMacEmitter, generate_background_macs, simulate_moving_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" # --------------------------------------------------------------------------- # Time anchors # --------------------------------------------------------------------------- WINDOW_OPEN = datetime(2025, 7, 3, 6, 0, 0, tzinfo=UTC) EEZ_CROSSING = datetime(2025, 7, 3, 7, 30, 0, tzinfo=UTC) TERRITORIAL_APPROACH= datetime(2025, 7, 3, 8, 30, 0, tzinfo=UTC) ANCHOR_TIME = datetime(2025, 7, 3, 9, 0, 0, tzinfo=UTC) CONTROLLER_BOOT = datetime(2025, 7, 3, 9, 15, 0, tzinfo=UTC) AIS_DARK_START = datetime(2025, 7, 3, 9, 25, 0, tzinfo=UTC) SWARM_LAUNCH = datetime(2025, 7, 3, 9, 30, 0, tzinfo=UTC) AIS_DARK_END = datetime(2025, 7, 3, 9, 35, 0, tzinfo=UTC) AIS_DARK2_START = datetime(2025, 7, 3, 9, 40, 0, tzinfo=UTC) AIS_DARK2_END = datetime(2025, 7, 3, 10, 15, 0, tzinfo=UTC) EEZ_EXIT = datetime(2025, 7, 3, 11, 30, 0, tzinfo=UTC) WINDOW_CLOSE = datetime(2025, 7, 3, 14, 0, 0, tzinfo=UTC) # --------------------------------------------------------------------------- # Geographic anchors # --------------------------------------------------------------------------- SHIP_LAT, SHIP_LON = 59.85, 24.70 # MV KASPIYSK anchor/hold position # MV KASPIYSK waypoints — strictly chronological KASPIYSK_WAYPOINTS: list[tuple[datetime, float, float]] = [ (WINDOW_OPEN, 59.40, 27.50), # entry, international waters (WINDOW_OPEN + timedelta(hours=1), 59.55, 26.80), # transit east GoF (EEZ_CROSSING, 59.62, 26.10), # Finnish EEZ crossing ~8 kn (TERRITORIAL_APPROACH, 59.75, 25.20), # approach slowing to 4 kn (ANCHOR_TIME, 59.85, 24.70), # anchor/hold position (AIS_DARK_START, 59.85, 24.70), # AIS goes dark (same position) # AIS dark window 1: 09:25–09:35 (drones launch at 09:30) (AIS_DARK_END, 59.85, 24.70), # brief AIS return # AIS dark window 2: 09:40–10:15 (AIS_DARK2_END, 59.83, 24.68), # AIS back, barely moved (AIS_DARK2_END + timedelta(minutes=45),59.55, 24.50), # departing south ~12 kn (AIS_DARK2_END + timedelta(hours=1, minutes=45), 59.20, 24.30), # exiting EEZ (WINDOW_CLOSE, 58.80, 24.00), # window close ] # AIS dark windows: two separate blackout periods AIS_DARK_WINDOWS = [ (AIS_DARK_START, AIS_DARK_END), (AIS_DARK2_START, AIS_DARK2_END), ] # --------------------------------------------------------------------------- # Drone swarm — 5 drones fan northwest toward Helsinki coast # --------------------------------------------------------------------------- # (track_id, target_lat, target_lon, alt_cruise_m, mac_addr, seed) SWARM: list[tuple[str, float, float, float, str, int]] = [ ("T-DRN-S8-01", 60.10, 24.50, 90.0, "5C:E2:8C:BB:01:01", 801), # Espoo coast ("T-DRN-S8-02", 60.15, 24.65, 95.0, "5C:E2:8C:BB:02:02", 802), # Lauttasaari ("T-DRN-S8-03", 60.18, 24.80, 100.0, "5C:E2:8C:BB:03:03", 803), # Helsinki West Harbor ("T-DRN-S8-04", 60.20, 24.95, 105.0, "5C:E2:8C:BB:04:04", 804), # Helsinki center coast ("T-DRN-S8-05", 60.17, 25.10, 85.0, "5C:E2:8C:BB:05:05", 805), # Kulosaari ] # Drone controller MAC (pre-launch smoking gun) CONTROLLER_MAC = "A4:83:E7:88:AA:01" CONTROLLER_BOOT_SENSOR = "MAC-HEL-COAST-01" # Patrol drone — Border Guard scrambles from Helsinki-Malmi at intercept bearing PAT_SCRAMBLE = SWARM_LAUNCH + timedelta(minutes=12) PAT_LAND = PAT_SCRAMBLE + timedelta(minutes=55) PATROL_PATH_3D: list[tuple[datetime, float, float, float]] = [ (PAT_SCRAMBLE, 60.254, 25.041, 0.0), (PAT_SCRAMBLE + timedelta(minutes=6), 60.200, 24.950, 210.0), (PAT_SCRAMBLE + timedelta(minutes=14), 60.150, 24.800, 230.0), (PAT_SCRAMBLE + timedelta(minutes=22), 60.100, 24.650, 225.0), # intercept swarm (PAT_SCRAMBLE + timedelta(minutes=32), 60.020, 24.700, 220.0), # approach ship (PAT_SCRAMBLE + timedelta(minutes=42), 59.870, 24.690, 195.0), # over mother ship (PAT_LAND, 60.254, 25.041, 0.0), ] PATROL_WAYPOINTS_2D = [(t, la, lo) for (t, la, lo, _) in PATROL_PATH_3D] PATROL_ALTITUDES = [(t, al) for (t, _la, _lo, al) in PATROL_PATH_3D] SENSORS_USED_IDS = { "MAC-HEL-COAST-01", "MAC-AIR-PLN-01", "MAC-AIR-DRN-01", "RAD-PLN-01", "RAD-DRN-PAT-01", } INFRA_USED_IDS = { "port-helsinki", "shipping-lane-eb", "shipping-lane-wb", } # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _drone_profile(swarm_entry: tuple): """Build 3D path and altitude profile for one swarm drone.""" track_id, tgt_lat, tgt_lon, alt_m, mac, seed = swarm_entry dist_m = haversine_m(SHIP_LAT, SHIP_LON, tgt_lat, tgt_lon) cruise_s = dist_m / 18.0 # 18 m/s cruise t0 = SWARM_LAUNCH t_cruise = t0 + timedelta(seconds=cruise_s) path_3d = [ (t0, SHIP_LAT, SHIP_LON, 0.0), (t0 + timedelta(seconds=30), SHIP_LAT, SHIP_LON, alt_m * 0.6), (t0 + timedelta(seconds=90), SHIP_LAT + (tgt_lat - SHIP_LAT) * 0.05, SHIP_LON + (tgt_lon - SHIP_LON) * 0.05, alt_m), (t_cruise - timedelta(seconds=30), tgt_lat - (tgt_lat - SHIP_LAT) * 0.02, tgt_lon - (tgt_lon - SHIP_LON) * 0.02, alt_m * 0.8), (t_cruise, tgt_lat, tgt_lon, 20.0), ] waypoints_2d = [(t, la, lo) for (t, la, lo, _) in path_3d] altitudes = [(t, al) for (t, _la, _lo, al) in path_3d] return path_3d, waypoints_2d, altitudes, track_id, mac, seed def build_ambient_ais(n_ships: int, seed: int) -> list[dict[str, Any]]: rng = random.Random(seed) out: list[dict[str, Any]] = [] for i in range(n_ships): eastbound = rng.random() < 0.5 lat0 = rng.uniform(59.20, 60.30) lat1 = lat0 + rng.uniform(-0.15, 0.15) lon0, lon1 = (22.5, 28.0) if eastbound else (28.0, 22.5) t_start = WINDOW_OPEN + timedelta(minutes=rng.uniform(0, 120)) t_end = min(WINDOW_CLOSE, t_start + timedelta(minutes=rng.uniform(90, 200))) 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=60.0, destination="FIHEL" if eastbound else "EETLL", seed=seed + i, ) out.extend(emit_ais(track)) return out def emit_controller_boot_macs(sensors: dict[str, Any]) -> list[MacObservation]: """Emit pre-launch controller + drone boot MAC observations from the coastal sensor MAC-HEL-COAST-01 at CONTROLLER_BOOT time (09:15 UTC). This is the smoking gun: controller and drone MACs detected 15 min before drones appear on radar.""" rng = random.Random(8801) out: list[MacObservation] = [] # Controller MAC: ~20 observations across the boot window ctrl_rssi_base = rng.uniform(-68.0, -72.0) for i in range(20): win_start = CONTROLLER_BOOT + timedelta(seconds=i * 28) win_end = win_start + timedelta(seconds=28) rssi = ctrl_rssi_base + rng.gauss(0, 1.5) out.append(MacObservation( sensor_id=CONTROLLER_BOOT_SENSOR, mac=CONTROLLER_MAC, session_start=win_start, session_end=win_end, message_count=max(1, int(rng.gauss(35, 8))), avg_rssi=round(rssi, 2), manufacturer="Apple", # controller iPad )) # Five drone MACs booting up — fewer obs per drone, lower RSSI (weaker tx during boot) for i, (track_id, _tgt_lat, _tgt_lon, _alt, mac, seed) in enumerate(SWARM): drone_rng = random.Random(8810 + i) boot_offset = timedelta(seconds=i * 15) # staggered boot n_obs = drone_rng.randint(2, 5) for j in range(n_obs): win_start = CONTROLLER_BOOT + boot_offset + timedelta(seconds=j * 45) win_end = win_start + timedelta(seconds=45) rssi = drone_rng.uniform(-73.0, -75.0) + drone_rng.gauss(0, 1.0) out.append(MacObservation( sensor_id=CONTROLLER_BOOT_SENSOR, mac=mac, session_start=win_start, session_end=win_end, message_count=max(1, int(drone_rng.gauss(12, 4))), avg_rssi=round(rssi, 2), manufacturer="DJI", )) return out # --------------------------------------------------------------------------- # Main generation functions # --------------------------------------------------------------------------- def generate_realtime() -> dict[str, int]: sensors = sensor_lookup() counts: dict[str, int] = {} # ----- AIS: MV KASPIYSK + ~400 ambient ships ----- kaspiysk_track = AisTrack( mmsi=636092744, waypoints=KASPIYSK_WAYPOINTS, cadence_s=60.0, dark_windows=AIS_DARK_WINDOWS, destination="FIHEL", nav_status=1, # at anchor during hold seed=801, ) ship_msgs = emit_ais(kaspiysk_track) ambient_msgs = build_ambient_ais(n_ships=400, seed=8000) ais_all = ship_msgs + ambient_msgs counts["ais.ndjson"] = write_ndjson( OUT_REALTIME / "ais.ndjson", ais_all, "s8-red-vessel/ais") snapshot_features = ais_snapshot_geojson(ais_all) counts["ais_snapshot.geojson"] = write_geojson( OUT_REALTIME / "ais_snapshot.geojson", snapshot_features, "s8-red-vessel/ais_snapshot") # ----- Drone radar: 5 swarm tracks + patrol drone ----- drone_recs: list[dict[str, Any]] = [] mac_obs: list[MacObservation] = [] for entry in SWARM: path_3d, wp2d, alts, track_id, mac_addr, entry_seed = _drone_profile(entry) dr_track = RadarTrack( track_id=track_id, sensor_id="RAD-PLN-01", waypoints=wp2d, cadence_s=2.0, classification="airborne_small", rcs_m2=0.025, confidence=0.72, seed=entry_seed, ) recs = emit_drone_radar(dr_track, alts) for r in recs: r["kind"] = "airborne" r["swarm_id"] = "SW-2025-0703-01" drone_recs.extend(recs) # MAC emission along drone path (coastal sensors pick up DJI MACs) em = MovingMacEmitter( mac=mac_addr, manufacturer="DJI", waypoints=wp2d, active_windows=[(wp2d[0][0], wp2d[-1][0])], seed=entry_seed + 1, ) mac_obs.extend(simulate_moving_mac(em, sensors, rssi_threshold=-100.0)) # Patrol drone track pat_track = RadarTrack( track_id="T-PAT-DRN-S8-01", sensor_id="RAD-DRN-PAT-01", waypoints=PATROL_WAYPOINTS_2D, cadence_s=3.0, classification="airborne_medium", rcs_m2=0.15, confidence=0.94, seed=899, ) pat_recs = emit_drone_radar(pat_track, PATROL_ALTITUDES) for r in pat_recs: r["kind"] = "patrol" drone_recs.extend(pat_recs) counts["drone_radar.ndjson"] = write_ndjson( OUT_REALTIME / "drone_radar.ndjson", drone_recs, "s8-red-vessel/drone_radar") # Plane radar: surface track of MV KASPIYSK plane_wp = [(t, la, lo) for (t, la, lo) in KASPIYSK_WAYPOINTS] plane_track = RadarTrack( track_id="T-PLN-S8-SURF-01", sensor_id="RAD-PLN-01", waypoints=plane_wp, cadence_s=4.0, classification="surface_large", rcs_m2=5200.0, confidence=0.92, seed=850, ) plane_recs = emit_radar(plane_track) for r in plane_recs: ts = datetime.strptime(r["timestamp"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=UTC) in_dark = any(d_start <= ts < d_end for d_start, d_end in AIS_DARK_WINDOWS) r["mmsi_hint"] = None if in_dark else 636092744 r["platform"] = "MAC-AIR-PLN-01" counts["plane_radar.ndjson"] = write_ndjson( OUT_REALTIME / "plane_radar.ndjson", plane_recs, "s8-red-vessel/plane_radar") # ----- MAC: controller boot (smoking gun) + drone MACs + background ----- # Controller boot observations — emitted before drones are airborne boot_obs = emit_controller_boot_macs(sensors) mac_obs.extend(boot_obs) bg = generate_background_macs(sensors, WINDOW_OPEN, WINDOW_CLOSE, mac_count=100, cadence_s=120.0, seed=42) 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, "s8-red-vessel/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, "s8-red-vessel/mac_sessions") # ----- Decimated companions ----- AIS_DECIM_FIELDS = ["timestamp", "lat", "lon", "sog_kn", "cog_deg", "nav_status"] DRONE_DECIM_FIELDS = ["timestamp", "lat", "lon", "alt_m", "speed_mps", "heading_deg", "rcs_m2", "classification", "kind", "swarm_id"] decim_reports = [] for path, kw in [ (OUT_REALTIME / "ais.ndjson", {"key_field": "mmsi", "ts_field": "ts_epoch_ms", "project_fields": AIS_DECIM_FIELDS}), (OUT_REALTIME / "drone_radar.ndjson", {"key_field": "track_id", "ts_field": "ts_epoch_ms", "project_fields": DRONE_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("[S8] 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 def generate_static() -> dict[str, int]: counts: dict[str, int] = {} # Area of interest: covers entry point to Helsinki coast (full 8h window) aoi = { "type": "Feature", "properties": { "featureId": "s8-aoi", "name": "S8 Area of Interest", "note": "MV KASPIYSK transit corridor + Helsinki coast drone fan", }, "geometry": {"type": "Polygon", "coordinates": [[ [24.00, 58.70], [28.00, 58.70], [28.00, 60.35], [24.00, 60.35], [24.00, 58.70], ]]}, } counts["area_of_interest.geojson"] = write_geojson( OUT_STATIC / "area_of_interest.geojson", [aoi], "s8-red-vessel/aoi") 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, "s8-red-vessel/sensors_used") 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, "s8-red-vessel/infra_used") return counts def generate_historical() -> dict[str, int]: counts: dict[str, int] = {} ais_baseline: list[dict[str, Any]] = [] baseline_days = [WINDOW_OPEN - timedelta(days=d) for d in range(1, 7)] for d in baseline_days: day_start = d.replace(hour=6, minute=0, second=0, microsecond=0) hist_wp = [ (day_start + timedelta(seconds=int((t - WINDOW_OPEN).total_seconds())), la, lo) for (t, la, lo) in KASPIYSK_WAYPOINTS ] # Keep only waypoints with non-negative time offsets (within same day window) hist_wp = [(t, la, lo) for (t, la, lo) in hist_wp if t >= day_start] if len(hist_wp) < 2: continue hist_track = AisTrack( mmsi=636092744, waypoints=hist_wp, cadence_s=60.0, destination="FIHEL", seed=9000 + d.day, ) ais_baseline.extend(emit_ais(hist_track)) counts["ais_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "ais_baseline.ndjson", ais_baseline, "s8-red-vessel/ais_baseline") 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: for d in [OUT_REALTIME, OUT_STATIC, OUT_HISTORICAL]: d.mkdir(parents=True, exist_ok=True) print("[S8] generating realtime layer …") rt = generate_realtime() print("[S8] generating static layer …") st = generate_static() print("[S8] generating historical layer …") hi = generate_historical() print("\n===== Scenario 08 — Red Vessel Escalation: generation summary =====") for section, data in [("realtime", rt), ("static", st), ("historical", hi)]: print(f"\n[{section}]") for k, v in data.items(): print(f" {k:<34} rows/features={v:>8}") rt_bytes = dir_size_bytes(OUT_REALTIME) print(f"\n[on disk] realtime {rt_bytes/1024/1024:.2f} MB") summary = { "scenario": "s8-red-vessel-escalation", "realtime": rt, "static": st, "historical": hi, } (SCENARIO_DIR / "data" / "_generation_summary.json").write_text( json.dumps(summary, indent=2), encoding="utf-8") print("[done] All files written under scenarios/08-red-vessel-escalation/data/") return 0 if __name__ == "__main__": raise SystemExit(main())