"""Scenario 05 — Drone Launch from Ship (Kilpilahti recon): data generator. Composes shared generators from `generators/` to produce realtime, static, and historical data for the S5 demo. See README.md for the full story. Pipeline: 1. MV AALLOTAR transits GoF westbound, loiters ~6.4 NM SSE of Emäsalo for ~49 min, then resumes. 2. Drone-MAC `5C:E2:8C:DD:EE:01` (DJI OUI) emits along a 3D path lifting from AALLOTAR (0 → 110 m), tracking 335° to the Kilpilahti recon polygon, executing a 14-min racetrack orbit, and returning to land on AALLOTAR. 3. Operator iPad MAC `A4:83:E7:5C:9B:10` rides AALLOTAR's waypoints (stays on the ship throughout). 4. Plane radar RAD-PLN-01 emits the surface-air picture, including the low-RCS airborne track `T-PLN-22871` for the suspect drone. 5. Border Guard patrol drone RAD-DRN-PAT-01 launches from Helsinki-Malmi, intercepts and shadows; carries airborne MAC sensor MAC-AIR-DRN-01, which captures BOTH the drone-MAC AND the operator-iPad-MAC during a slow pass over AALLOTAR — the smoking-gun co-observation. 6. Coastal sensors MAC-PRV-COAST-02 / -01 and MAC-HEL-PORT-03 acquire the drone-MAC during overflight (custom high-tx-power LOS propagation, because consumer-grade indoor path-loss n=2.2 with tx=-15 dBm cannot deliver -61 dBm at 25 km). 7. Decoy: coastguard inspection drone `90:3A:E6:11:22:33` (Parrot OUI) operates the same polygon earlier the same day with filed flight plan `FP-CG-2025-0418-007`; must NOT trigger. 8. Historical: 8 weeks ambient AIS + ~30 legitimate near-shore drone flights written to `data/historical/legitimate_drones.ndjson`. All shared logic comes from `generators/`; scenario-specific helpers (airborne-drone MAC emission at long range, decoy track synthesis, flight plan registry) are kept local to this file. """ 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, crew_by_ship, haversine_m, iso_utc, load_infrastructure, load_personas, 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_airborne_mac, simulate_moving_mac, ) from generators.radar_generator import RadarTrack, emit_drone_radar, emit_radar, surface_to_air_record # 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 (spec §2 Timeline) # --------------------------------------------------------------------------- WINDOW_OPEN = datetime(2025, 4, 18, 5, 0, 0, tzinfo=UTC) WINDOW_CLOSE = datetime(2025, 4, 18, 15, 0, 0, tzinfo=UTC) # Decoy timeline DECOY_LAUNCH = datetime(2025, 4, 18, 6, 12, 0, tzinfo=UTC) DECOY_ORBIT_START = datetime(2025, 4, 18, 6, 35, 0, tzinfo=UTC) DECOY_ORBIT_END = datetime(2025, 4, 18, 6, 44, 0, tzinfo=UTC) DECOY_LAND = datetime(2025, 4, 18, 7, 5, 0, tzinfo=UTC) # AALLOTAR transit AAL_T1 = datetime(2025, 4, 18, 11, 48, 30, tzinfo=UTC) AAL_T2 = datetime(2025, 4, 18, 12, 18, 0, tzinfo=UTC) AAL_T3 = datetime(2025, 4, 18, 12, 31, 0, tzinfo=UTC) LOITER_T0 = datetime(2025, 4, 18, 12, 46, 10, tzinfo=UTC) LOITER_T1 = datetime(2025, 4, 18, 13, 32, 0, tzinfo=UTC) AAL_RESUME = datetime(2025, 4, 18, 13, 35, 0, tzinfo=UTC) AAL_T7 = datetime(2025, 4, 18, 14, 5, 0, tzinfo=UTC) # Suspect drone airborne window DRONE_TAKEOFF = datetime(2025, 4, 18, 12, 58, 0, tzinfo=UTC) DRONE_POLY_ENTRY = datetime(2025, 4, 18, 13, 9, 20, tzinfo=UTC) DRONE_ORBIT_START = datetime(2025, 4, 18, 13, 10, 0, tzinfo=UTC) DRONE_ORBIT_END = datetime(2025, 4, 18, 13, 23, 40, tzinfo=UTC) DRONE_LAND = datetime(2025, 4, 18, 13, 34, 18, tzinfo=UTC) # Patrol drone (RAD-DRN-PAT-01) timeline PAT_LAUNCH = datetime(2025, 4, 18, 13, 14, 30, tzinfo=UTC) PAT_COOBS_1 = datetime(2025, 4, 18, 13, 32, 7, tzinfo=UTC) PAT_COOBS_2 = datetime(2025, 4, 18, 13, 32, 42, tzinfo=UTC) PAT_LAND = datetime(2025, 4, 18, 13, 40, 0, tzinfo=UTC) # --------------------------------------------------------------------------- # Geographic anchors (spec §3 Geographic Layout) # --------------------------------------------------------------------------- # AALLOTAR waypoints: (t, lat, lon) AAL_WAYPOINTS: list[tuple[datetime, float, float]] = [ (AAL_T1, 60.0420, 25.9200), (AAL_T2, 60.0780, 25.8050), (AAL_T3, 60.1010, 25.7220), (LOITER_T0, 60.1130, 25.6520), # W4 loiter on-station (LOITER_T1, 60.1135, 25.6510), # W5 still on-station (~0.6 kn drift) (AAL_RESUME, 60.1140, 25.6480), # W6 spool up (AAL_T7, 60.1280, 25.5100), # W7 westbound resume ] LOITER_LAT, LOITER_LON = 60.1130, 25.6520 # Suspect drone 3D path: (t, lat, lon, alt_m) DRONE_PATH_3D: list[tuple[datetime, float, float, float]] = [ (DRONE_TAKEOFF, 60.1130, 25.6520, 0.0), (datetime(2025, 4, 18, 13, 0, 30, tzinfo=UTC), 60.1230, 25.6470, 110.0), (datetime(2025, 4, 18, 13, 4, 12, tzinfo=UTC), 60.1640, 25.6360, 112.0), (datetime(2025, 4, 18, 13, 7, 55, tzinfo=UTC), 60.2900, 25.5360, 115.0), (DRONE_POLY_ENTRY, 60.3010, 25.5180, 108.0), (DRONE_ORBIT_START, 60.3045, 25.5135, 108.0), (datetime(2025, 4, 18, 13, 13, 0, tzinfo=UTC), 60.3120, 25.5310, 110.0), (datetime(2025, 4, 18, 13, 16, 0, tzinfo=UTC), 60.3060, 25.5520, 109.0), (datetime(2025, 4, 18, 13, 19, 0, tzinfo=UTC), 60.3000, 25.5305, 108.0), (datetime(2025, 4, 18, 13, 22, 0, tzinfo=UTC), 60.3070, 25.5145, 110.0), (DRONE_ORBIT_END, 60.3040, 25.5170, 109.0), (datetime(2025, 4, 18, 13, 27, 10, tzinfo=UTC), 60.2880, 25.5410, 90.0), (datetime(2025, 4, 18, 13, 31, 0, tzinfo=UTC), 60.1500, 25.6280, 60.0), (DRONE_LAND, 60.1130, 25.6520, 0.0), ] DRONE_WAYPOINTS_2D: list[tuple[datetime, float, float]] = [ (t, lat, lon) for (t, lat, lon, _alt) in DRONE_PATH_3D ] DRONE_ALTITUDES: list[tuple[datetime, float]] = [ (t, alt) for (t, _lat, _lon, alt) in DRONE_PATH_3D ] # Patrol drone path (RAD-DRN-PAT-01, carrying MAC-AIR-DRN-01): (t, lat, lon, alt_m) PATROL_PATH_3D: list[tuple[datetime, float, float, float]] = [ (PAT_LAUNCH, 60.2540, 25.0410, 0.0), (datetime(2025, 4, 18, 13, 18, 0, tzinfo=UTC), 60.2200, 25.2200, 240.0), (datetime(2025, 4, 18, 13, 24, 0, tzinfo=UTC), 60.1700, 25.4500, 240.0), (datetime(2025, 4, 18, 13, 30, 0, tzinfo=UTC), 60.1280, 25.6100, 230.0), (PAT_COOBS_1, 60.1132, 25.6515, 220.0), (PAT_COOBS_2, 60.1145, 25.6540, 220.0), (PAT_LAND, 60.1900, 25.4000, 240.0), ] PATROL_WAYPOINTS_2D: list[tuple[datetime, float, float]] = [ (t, lat, lon) for (t, lat, lon, _alt) in PATROL_PATH_3D ] PATROL_ALTITUDES: list[tuple[datetime, float]] = [ (t, alt) for (t, _lat, _lon, alt) in PATROL_PATH_3D ] # Kilpilahti recon polygon (spec §3); also annotated as `purpose=recon_target` KILPILAHTI_POLYGON = [ [25.5050, 60.3120], [25.5640, 60.3120], [25.5640, 60.2950], [25.5050, 60.2950], [25.5050, 60.3120], ] KILPILAHTI_CLAT = sum(p[1] for p in KILPILAHTI_POLYGON[:-1]) / 4.0 KILPILAHTI_CLON = sum(p[0] for p in KILPILAHTI_POLYGON[:-1]) / 4.0 # Sköldvik RPAS base (decoy launch site, near Kilpilahti shore) SKOLDVIK_LAT, SKOLDVIK_LON = 60.3060, 25.5350 # Decoy drone 3D path (Parrot UAV; shore launch, polygon orbit, RTB) DECOY_PATH_3D: list[tuple[datetime, float, float, float]] = [ (DECOY_LAUNCH, SKOLDVIK_LAT, SKOLDVIK_LON, 0.0), (DECOY_LAUNCH + timedelta(minutes=8), 60.3070, 25.5300, 90.0), (DECOY_ORBIT_START, 60.3045, 25.5135, 95.0), (datetime(2025, 4, 18, 6, 38, 0, tzinfo=UTC), 60.3110, 25.5330, 95.0), (datetime(2025, 4, 18, 6, 41, 0, tzinfo=UTC), 60.3050, 25.5520, 95.0), (DECOY_ORBIT_END, 60.3035, 25.5345, 95.0), (datetime(2025, 4, 18, 6, 55, 0, tzinfo=UTC), 60.3070, 25.5340, 70.0), (DECOY_LAND, SKOLDVIK_LAT, SKOLDVIK_LON, 0.0), ] DECOY_WAYPOINTS_2D: list[tuple[datetime, float, float]] = [ (t, lat, lon) for (t, lat, lon, _alt) in DECOY_PATH_3D ] DECOY_ALTITUDES: list[tuple[datetime, float]] = [ (t, alt) for (t, _lat, _lon, alt) in DECOY_PATH_3D ] # Spec MAC IDs DRONE_MAC = "5C:E2:8C:DD:EE:01" # DJI OUI (suspect) OPERATOR_IPAD_MAC = "A4:83:E7:5C:9B:10" # P-AAL-OP1 iPad DECOY_DRONE_MAC = "90:3A:E6:11:22:33" # Parrot OUI (decoy) # Sensors with line-of-sight to drone overflight (per spec) COASTAL_DRONE_SENSORS = ["MAC-PRV-COAST-02", "MAC-PRV-COAST-01", "MAC-HEL-PORT-03"] # Sensor subset used in scenario SENSORS_USED_IDS = { "MAC-PRV-COAST-02", "MAC-PRV-COAST-01", "MAC-HEL-PORT-03", "MAC-HEL-COAST-01", "MAC-HEL-PORT-01", "MAC-AIR-DRN-01", "MAC-AIR-PLN-01", "RAD-PLN-01", "RAD-DRN-PAT-01", "RAD-COAST-HEL-01", } # Infrastructure subset (catalog featureIds) — used in static layer INFRA_USED_IDS = { "port-helsinki", "port-kilpilahti", "port-porvoo", "site-vuosaari", "finnish-eez-gof", "shipping-lane-wb", "shipping-lane-eb", } # --------------------------------------------------------------------------- # Custom airborne-MAC propagation for the suspect drone. # # The shared `simulate_moving_mac` uses rssi_from_distance(tx_dbm=-15, n=2.2) # which is calibrated for ship-mounted devices reaching coastal sensors at # 5-18 km. The suspect drone is an airborne DJI Wi-Fi emitter at 100+ m # altitude, and the closest coastal sensor (MAC-PRV-COAST-02) is ~26 km from # the polygon centroid — we need a free-space LOS model (n=2.0) with a much # higher effective tx_dbm (~+22 dBm, achievable for 100 mW Wi-Fi with antenna # gain) to land RSSI in the spec band of -61 .. -84 dBm. # --------------------------------------------------------------------------- def emit_drone_mac_at_coastal_sensors( waypoints_2d: list[tuple[datetime, float, float]], mac: str, manufacturer: str, active_window: tuple[datetime, datetime], sensor_ids: list[str], sensors: dict[str, dict[str, Any]], *, tx_dbm: float = 22.0, n: float = 2.0, noise_db: float = 2.5, session_window_s: float = 30.0, rssi_threshold: float = -90.0, seed: int = 5101, ) -> list[MacObservation]: """Walk the drone's 3D path in `session_window_s` slices; for each listed coastal sensor compute the RSSI using free-space LOS with elevated tx power and emit a `MacObservation` when above threshold.""" rng = random.Random(seed) out: list[MacObservation] = [] win_start, win_end = active_window t = win_start while t < win_end: seg = (t, min(win_end, t + timedelta(seconds=session_window_s))) tm = seg[0] + (seg[1] - seg[0]) / 2 lat, lon = _track_pos_2d(waypoints_2d, tm) for sid in sensor_ids: s = sensors.get(sid) if not s: continue d = haversine_m(s["lat"], s["lon"], lat, lon) rssi = tx_dbm - 10.0 * n * math.log10(max(d, 1.0)) + rng.gauss(0.0, noise_db) if rssi < rssi_threshold: continue # Occasional propagation dropouts at the very faint end if rssi < -82.0 and rng.random() < 0.5: continue msg_count = max(1, int(rng.gauss(6, 2))) out.append(MacObservation( sensor_id=sid, mac=mac, session_start=seg[0], session_end=seg[1], message_count=msg_count, avg_rssi=rssi, manufacturer=manufacturer, version="1.4", status="ACTIVE" if rssi > -80.0 else "FRINGE", )) t = seg[1] return out def _track_pos_2d(track: list[tuple[datetime, float, float]], t: datetime) -> tuple[float, float]: if t <= track[0][0]: return track[0][1], track[0][2] if t >= track[-1][0]: return track[-1][1], track[-1][2] for i in range(len(track) - 1): a, b = track[i], track[i + 1] if a[0] <= t <= b[0]: span = (b[0] - a[0]).total_seconds() f = 0.0 if span <= 0 else (t - a[0]).total_seconds() / span return a[1] + f * (b[1] - a[1]), a[2] + f * (b[2] - a[2]) return track[-1][1], track[-1][2] def _alt_at(alts: list[tuple[datetime, float]], t: datetime) -> float: if t <= alts[0][0]: return alts[0][1] if t >= alts[-1][0]: return alts[-1][1] for i in range(len(alts) - 1): a, b = alts[i], alts[i + 1] if a[0] <= t <= b[0]: span = (b[0] - a[0]).total_seconds() f = 0.0 if span <= 0 else (t - a[0]).total_seconds() / span return a[1] + f * (b[1] - a[1]) return alts[-1][1] def emit_airborne_drone_mac_boosted( sensor_track_2d: list[tuple[datetime, float, float]], drone_waypoints_2d: list[tuple[datetime, float, float]], drone_altitudes: list[tuple[datetime, float]], patrol_altitudes: list[tuple[datetime, float]], mac: str, manufacturer: str, active_window: tuple[datetime, datetime], *, tx_dbm: float = 12.0, n: float = 1.9, noise_db: float = 2.5, session_window_s: float = 5.0, rssi_threshold: float = -90.0, seed: int = 5503, ) -> list[MacObservation]: """Airborne MAC capture of the suspect drone with elevated effective tx power (models DJI directional antenna gain seen by near-overhead patrol sensor) and 3D distance including altitude. The spec narrates a near-overhead pass producing RSSI ≈ -58 dBm at the co-obs moments — consistent with this LOS model when the patrol drone is over AALLOTAR and the suspect is still hovering 1-2 km out at low altitude.""" rng = random.Random(seed) win_start, win_end = active_window t = win_start t_first_sensor, t_last_sensor = sensor_track_2d[0][0], sensor_track_2d[-1][0] out: list[MacObservation] = [] while t < win_end: seg = (t, min(win_end, t + timedelta(seconds=session_window_s))) tm = seg[0] + (seg[1] - seg[0]) / 2 # Skip windows where the patrol sensor is not yet airborne if tm < t_first_sensor or tm > t_last_sensor: t = seg[1] continue slat, slon = _track_pos_2d(sensor_track_2d, tm) salt = _alt_at(patrol_altitudes, tm) dlat, dlon = _track_pos_2d(drone_waypoints_2d, tm) dalt = _alt_at(drone_altitudes, tm) horiz = haversine_m(slat, slon, dlat, dlon) vert = abs(salt - dalt) d = math.sqrt(horiz * horiz + vert * vert) rssi = tx_dbm - 10.0 * n * math.log10(max(d, 1.0)) + rng.gauss(0.0, noise_db) if rssi < rssi_threshold: t = seg[1] continue out.append(MacObservation( sensor_id="MAC-AIR-DRN-01", mac=mac, session_start=seg[0], session_end=seg[1], message_count=max(1, int(rng.gauss(9, 3))), avg_rssi=rssi, manufacturer=manufacturer, version="1.4", status="ACTIVE", )) t = seg[1] return out # --------------------------------------------------------------------------- # Background AIS in the GoF AOI # --------------------------------------------------------------------------- def build_ambient_ais(n_ships: int, start: datetime, end: datetime, 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.65, 60.30) lat1 = lat0 + rng.uniform(-0.10, 0.10) lon0, lon1 = (22.5, 27.5) if eastbound else (27.5, 22.5) day_offset = rng.uniform(0.0, max(1.0, (end - start).total_seconds() - 6 * 3600)) t_start = start + timedelta(seconds=day_offset) t_end = t_start + timedelta(minutes=rng.uniform(70, 160)) if t_end > end: 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 # --------------------------------------------------------------------------- # Realtime layer # --------------------------------------------------------------------------- def generate_realtime() -> dict[str, int]: sensors = sensor_lookup() counts: dict[str, int] = {} # ----- AIS — AALLOTAR ----- aal_track = AisTrack( mmsi=230999401, waypoints=AAL_WAYPOINTS, cadence_s=10.0, destination="FIHEL", nav_status=0, speed_jitter_kn=0.3, course_jitter_deg=2.0, seed=5001, ) aal_msgs = emit_ais(aal_track) # ----- AIS — ambient GoF traffic (scenario window only) ----- ambient_msgs = build_ambient_ais(n_ships=1400, start=WINDOW_OPEN, end=WINDOW_CLOSE, seed=5002) ais_all = aal_msgs + ambient_msgs counts["ais.ndjson"] = write_ndjson( OUT_REALTIME / "ais.ndjson", ais_all, "s5-drone-launch-from-ship/ais") counts["ais_snapshot.geojson"] = write_geojson( OUT_REALTIME / "ais_snapshot.geojson", ais_snapshot_geojson(ais_all), "s5-drone-launch-from-ship/ais_snapshot") # ----- Plane radar RAD-PLN-01: suspect drone airborne track T-PLN-22871 ----- suspect_drone_radar = RadarTrack( track_id="T-PLN-22871", sensor_id="RAD-PLN-01", waypoints=DRONE_WAYPOINTS_2D, cadence_s=2.0, classification="air_small", rcs_m2=0.03, confidence=0.66, seed=5101, ) plane_radar_msgs = emit_drone_radar(suspect_drone_radar, altitudes=DRONE_ALTITUDES) # Annotate spec metadata for r in plane_radar_msgs: r["kind"] = "airborne" r["flight_plan_id"] = None r["origin"] = "offshore" r["mmsi_hint"] = 230999401 # spatially co-located with AALLOTAR # Decoy drone earlier in the day, observed by RAD-PLN-01 (surface-air picture) decoy_plane_radar = RadarTrack( track_id="T-PLN-22810", sensor_id="RAD-PLN-01", waypoints=DECOY_WAYPOINTS_2D, cadence_s=2.0, classification="air_small", rcs_m2=0.06, confidence=0.74, seed=5102, ) plane_radar_decoy = emit_drone_radar(decoy_plane_radar, altitudes=DECOY_ALTITUDES) for r in plane_radar_decoy: r["kind"] = "airborne" r["flight_plan_id"] = "FP-CG-2025-0418-007" r["origin"] = "shore" # Plane radar also routinely sees AALLOTAR as a surface contact plane_radar_surface = emit_radar(RadarTrack( track_id="T-PLN-S-44912", sensor_id="RAD-PLN-01", waypoints=[(t, lat, lon) for (t, lat, lon) in AAL_WAYPOINTS], cadence_s=8.0, classification="surface_large", rcs_m2=8200.0, confidence=0.92, seed=5103, )) for r in plane_radar_surface: r["kind"] = "surface" r["mmsi_hint"] = 230999401 plane_radar_all = plane_radar_msgs + plane_radar_decoy + plane_radar_surface counts["plane_radar.ndjson"] = write_ndjson( OUT_REALTIME / "plane_radar.ndjson", plane_radar_all, "s5-drone-launch-from-ship/plane_radar") # ----- Drone radar RAD-DRN-PAT-01: patrol drone's own nose radar plots ----- # The patrol drone tracks AALLOTAR (surface) and the suspect drone (air). patrol_radar_surface = emit_radar(RadarTrack( track_id="T-DRN-771", sensor_id="RAD-DRN-PAT-01", waypoints=[(t, LOITER_LAT + 0.0005 * math.sin(i), LOITER_LON + 0.0005 * math.cos(i)) for i, t in enumerate([ datetime(2025, 4, 18, 13, 30, 0, tzinfo=UTC), datetime(2025, 4, 18, 13, 31, 0, tzinfo=UTC), datetime(2025, 4, 18, 13, 32, 0, tzinfo=UTC), datetime(2025, 4, 18, 13, 33, 0, tzinfo=UTC), datetime(2025, 4, 18, 13, 34, 0, tzinfo=UTC), ])], cadence_s=1.0, classification="surface_large", rcs_m2=7800.0, confidence=0.94, seed=5201, )) for r in patrol_radar_surface: r["kind"] = "surface" r["mmsi_hint"] = 230999401 surface_to_air_record(r) # drone_radar DDL: alt_m, speed_mps, heading_deg # Patrol drone's own air-track of the suspect drone during the final approach patrol_radar_air_window = [ (t, lat, lon, alt) for (t, lat, lon, alt) in DRONE_PATH_3D if PAT_LAUNCH <= t <= PAT_LAND ] if len(patrol_radar_air_window) >= 2: patrol_radar_air = emit_drone_radar(RadarTrack( track_id="T-DRN-772", sensor_id="RAD-DRN-PAT-01", waypoints=[(t, lat, lon) for (t, lat, lon, _alt) in patrol_radar_air_window], cadence_s=1.0, classification="air_small", rcs_m2=0.03, confidence=0.71, seed=5202, ), altitudes=[(t, alt) for (t, _lat, _lon, alt) in patrol_radar_air_window]) for r in patrol_radar_air: r["kind"] = "airborne" else: patrol_radar_air = [] drone_radar_all = patrol_radar_surface + patrol_radar_air counts["drone_radar.ndjson"] = write_ndjson( OUT_REALTIME / "drone_radar.ndjson", drone_radar_all, "s5-drone-launch-from-ship/drone_radar") # ----- MAC stream ----- mac_obs: list[MacObservation] = [] # AALLOTAR crew (persistent MACs) ride the ship. Active for the full window. aal_active = [(WINDOW_OPEN, WINDOW_CLOSE)] for i, p in enumerate(crew_by_ship("MV AALLOTAR")): if not p.get("persistent"): continue em = MovingMacEmitter( mac=p["device_mac"], manufacturer=p["oui_vendor"], waypoints=AAL_WAYPOINTS, active_windows=aal_active, seed=5300 + i, ) # NOTE: the operator iPad A4:83:E7:5C:9B:10 (P-AAL-OP1) is one of the # crew MACs and is emitted here via MovingMacEmitter riding AALLOTAR. mac_obs.extend(simulate_moving_mac(em, sensors, session_window_s=180.0)) # Suspect drone MAC observed at the three coastal sensors during the # airborne window. Custom emission tuned for airborne Wi-Fi LOS over # 20-40 km (see emit_drone_mac_at_coastal_sensors docstring). mac_obs.extend(emit_drone_mac_at_coastal_sensors( waypoints_2d=DRONE_WAYPOINTS_2D, mac=DRONE_MAC, manufacturer="DJI", active_window=(DRONE_TAKEOFF, DRONE_LAND), sensor_ids=COASTAL_DRONE_SENSORS, sensors=sensors, session_window_s=30.0, seed=5401, )) # Decoy coastguard drone MAC during decoy window — observed at the same # coastal sensors; status="ALLOWLIST" so the fusion gate suppresses it. decoy_coast_obs = emit_drone_mac_at_coastal_sensors( waypoints_2d=DECOY_WAYPOINTS_2D, mac=DECOY_DRONE_MAC, manufacturer="Parrot", active_window=(DECOY_LAUNCH, DECOY_LAND), sensor_ids=COASTAL_DRONE_SENSORS, sensors=sensors, session_window_s=60.0, seed=5402, ) for ob in decoy_coast_obs: ob.status = "ALLOWLIST" mac_obs.extend(decoy_coast_obs) # Airborne MAC sensor MAC-AIR-DRN-01 on the patrol drone. Targets: # both the suspect drone MAC and the operator iPad MAC. ipad_mac_emitter = MovingMacEmitter( mac=OPERATOR_IPAD_MAC, manufacturer="Apple", waypoints=AAL_WAYPOINTS, active_windows=[(WINDOW_OPEN, WINDOW_CLOSE)], seed=5502, ) # iPad is on AALLOTAR (small distance to patrol drone over-the-top pass); # default propagation produces RSSI in the spec's -54 dBm band. airborne_obs_ipad = simulate_airborne_mac( airborne_sensor_id="MAC-AIR-DRN-01", sensor_track=PATROL_WAYPOINTS_2D, target_macs=[ipad_mac_emitter], rssi_threshold=-90.0, session_window_s=5.0, ) # Drone MAC: the spec's drone-path geometry places D13/D14 along the # egress leg, so the drone is still ~2-3 km from the patrol drone at the # co-obs moment. To deliver the spec-narrated airborne RSSI of -58 dBm # (consistent with a DJI Wi-Fi emitter with a directional antenna seen # by a near-overhead receiver), we emit the airborne drone-MAC # observations with an elevated effective tx_dbm and LOS path loss. airborne_obs_drone = emit_airborne_drone_mac_boosted( sensor_track_2d=PATROL_WAYPOINTS_2D, drone_waypoints_2d=DRONE_WAYPOINTS_2D, drone_altitudes=DRONE_ALTITUDES, patrol_altitudes=PATROL_ALTITUDES, mac=DRONE_MAC, manufacturer="DJI", active_window=(DRONE_TAKEOFF, DRONE_LAND), session_window_s=5.0, seed=5503, ) airborne_obs = airborne_obs_ipad + airborne_obs_drone mac_obs.extend(airborne_obs) # Background consumer MACs across the realtime window bg = generate_background_macs( sensors, WINDOW_OPEN, WINDOW_CLOSE, mac_count=30, cadence_s=600.0, seed=5601, ) 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, "s5-drone-launch-from-ship/mac") counts["mac.csv"] = write_csv( OUT_REALTIME / "mac.csv", MAC_CSV_HEADER, [m.to_csv_row() for m in mac_obs], "s5-drone-launch-from-ship/mac_sessions") # Sanity: did we actually produce the dual-MAC co-observation? cooobs_window_macs: set[str] = set() for ob in airborne_obs: if PAT_COOBS_1 - timedelta(seconds=10) <= ob.session_start <= PAT_COOBS_2 + timedelta(seconds=10): cooobs_window_macs.add(ob.mac.lower()) counts["__cooobs_macs"] = len(cooobs_window_macs) if DRONE_MAC.lower() not in cooobs_window_macs or OPERATOR_IPAD_MAC.lower() not in cooobs_window_macs: print(f"[WARN] expected dual-MAC co-observation at MAC-AIR-DRN-01 in window " f"{PAT_COOBS_1.isoformat()}..{PAT_COOBS_2.isoformat()}; got {cooobs_window_macs}") else: print(f"[OK] dual-MAC co-observation confirmed: {sorted(cooobs_window_macs)}") 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}), ]: 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("[S5] 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 GeoJSON layer # --------------------------------------------------------------------------- def _circle_polygon(clat: float, clon: float, radius_m: float, n: int = 64) -> list[list[float]]: pts = [] m_per_deg_lat = 111_000.0 m_per_deg_lon = 111_000.0 * math.cos(math.radians(clat)) for i in range(n + 1): a = 2.0 * math.pi * i / n dx = radius_m * math.cos(a) dy = radius_m * math.sin(a) pts.append([clon + dx / m_per_deg_lon, clat + dy / m_per_deg_lat]) return pts def _synthetic_south_coastline() -> list[list[list[float]]]: """Coarse MultiLineString approximation of the Finnish southern coastline between Hanko and Kotka. Synthetic (looking-real, not surveyed).""" # Two segments: Hanko -> Helsinki and Helsinki -> Kotka, sampled coarsely. seg_a = [ [22.97, 59.83], [23.40, 59.86], [23.90, 59.92], [24.40, 59.99], [24.70, 60.05], [24.92, 60.10], [25.05, 60.16], ] seg_b = [ [25.05, 60.16], [25.30, 60.21], [25.55, 60.30], [25.75, 60.34], [26.20, 60.40], [26.60, 60.43], [26.95, 60.46], ] return [seg_a, seg_b] def generate_static() -> dict[str, int]: counts: dict[str, int] = {} # Area of interest aoi = { "type": "Feature", "properties": { "featureId": "s5-aoi", "name": "S5 Area of Interest", "note": "Bounding polygon covering AALLOTAR transit, drone 3D path, and patrol drone egress", }, "geometry": {"type": "Polygon", "coordinates": [[ [24.90, 59.95], [26.10, 59.95], [26.10, 60.45], [24.90, 60.45], [24.90, 59.95], ]]}, } counts["area_of_interest.geojson"] = write_geojson( OUT_STATIC / "area_of_interest.geojson", [aoi], "s5-drone-launch-from-ship/area_of_interest") # Catalog subsets 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, "s5-drone-launch-from-ship/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, "s5-drone-launch-from-ship/infrastructure_used") # Kilpilahti recon polygon (spec) recon_feat = { "type": "Feature", "properties": { "featureId": "kilpilahti-recon-polygon", "name": "Kilpilahti recon polygon (S5)", "purpose": "recon_target", "note": "Spec §3 vertices; geographically correct Kilpilahti location", }, "geometry": {"type": "Polygon", "coordinates": [KILPILAHTI_POLYGON]}, } counts["kilpilahti_recon_polygon.geojson"] = write_geojson( OUT_STATIC / "kilpilahti_recon_polygon.geojson", [recon_feat], "s5-drone-launch-from-ship/kilpilahti_recon_polygon") # Finnish southern coastline (synthetic MultiLineString) coast_feat = { "type": "Feature", "properties": { "featureId": "finnish-coastline-south", "name": "Finnish coastline (south) — synthetic approximation", "note": "Coarse MultiLineString sampled between Hanko and Kotka", }, "geometry": {"type": "MultiLineString", "coordinates": _synthetic_south_coastline()}, } counts["finnish_coastline_south.geojson"] = write_geojson( OUT_STATIC / "finnish_coastline_south.geojson", [coast_feat], "s5-drone-launch-from-ship/finnish_coastline_south") # 12 NM territorial-sea buffer (synthetic simple buffer south of coastline) # We render a polygon south of the coastline by offsetting by ~12 NM = 22.2 km # to the south for each coastline vertex; closed back along the coastline. one_nm_deg_lat = 1.0 / 60.0 twelve_nm_lat = -12.0 * one_nm_deg_lat south_pts = [] for seg in _synthetic_south_coastline(): for lon, lat in seg: south_pts.append([lon, lat + twelve_nm_lat]) north_pts = [] for seg in reversed(_synthetic_south_coastline()): for lon, lat in reversed(seg): north_pts.append([lon, lat]) ring = south_pts + north_pts ring.append(ring[0]) ts_feat = { "type": "Feature", "properties": { "featureId": "fi-territorial-sea-12nm", "name": "Finnish 12 NM territorial sea (synthetic buffer south of coastline)", "note": "Simplified buffer used for offshore_km classification", }, "geometry": {"type": "Polygon", "coordinates": [ring]}, } counts["fi_territorial_sea_12nm.geojson"] = write_geojson( OUT_STATIC / "fi_territorial_sea_12nm.geojson", [ts_feat], "s5-drone-launch-from-ship/fi_territorial_sea_12nm") # AALLOTAR track (LineString with per-vertex timestamps) aal_feat = { "type": "Feature", "properties": { "featureId": "s5-aallotar-track", "name": "MV AALLOTAR W1..W7 track", "mmsi": 230999401, "timestamps": [t.isoformat() for (t, _lat, _lon) in AAL_WAYPOINTS], }, "geometry": {"type": "LineString", "coordinates": [[lon, lat] for (_t, lat, lon) in AAL_WAYPOINTS]}, } counts["s5_aallotar_track.geojson"] = write_geojson( OUT_STATIC / "s5_aallotar_track.geojson", [aal_feat], "s5-drone-launch-from-ship/s5_aallotar_track") # Suspect drone 3D path (LineString with per-vertex altitudes) drone_feat = { "type": "Feature", "properties": { "featureId": "s5-drone-path-3d", "name": "Suspect drone D1..D14 3D path", "mac": DRONE_MAC, "altitudes_m": [alt for (_t, _lat, _lon, alt) in DRONE_PATH_3D], "timestamps": [t.isoformat() for (t, _lat, _lon, _alt) in DRONE_PATH_3D], }, "geometry": {"type": "LineString", "coordinates": [[lon, lat, alt] for (_t, lat, lon, alt) in DRONE_PATH_3D]}, } counts["s5_drone_path_3d.geojson"] = write_geojson( OUT_STATIC / "s5_drone_path_3d.geojson", [drone_feat], "s5-drone-launch-from-ship/s5_drone_path_3d") # Patrol drone path pat_feat = { "type": "Feature", "properties": { "featureId": "s5-patrol-drone-path", "name": "RAD-DRN-PAT-01 patrol drone P1..P7 path", "timestamps": [t.isoformat() for (t, _lat, _lon, _alt) in PATROL_PATH_3D], "altitudes_m": [alt for (_t, _lat, _lon, alt) in PATROL_PATH_3D], }, "geometry": {"type": "LineString", "coordinates": [[lon, lat, alt] for (_t, lat, lon, alt) in PATROL_PATH_3D]}, } counts["s5_patrol_drone_path.geojson"] = write_geojson( OUT_STATIC / "s5_patrol_drone_path.geojson", [pat_feat], "s5-drone-launch-from-ship/s5_patrol_drone_path") # Sköldvik RPAS base (decoy launch site) rpas_feat = { "type": "Feature", "properties": { "featureId": "rpas-base-skoldvik", "name": "Sköldvik RPAS base (decoy shore launch)", "operator": "Finnish Border Guard - coastguard inspection UAS", }, "geometry": {"type": "Point", "coordinates": [SKOLDVIK_LON, SKOLDVIK_LAT]}, } counts["rpas_base_skoldvik.geojson"] = write_geojson( OUT_STATIC / "rpas_base_skoldvik.geojson", [rpas_feat], "s5-drone-launch-from-ship/rpas_base_skoldvik") # Decoy flight-plan registry entry (geometry covers Kilpilahti polygon) fp_feat = { "type": "Feature", "properties": { "featureId": "decoy-flight-plan", "flight_plan_id": "FP-CG-2025-0418-007", "operator": "Finnish Border Guard", "asset": "ASSET-DRN-CG-INS-04", "mac": DECOY_DRONE_MAC, "oui_vendor": "Parrot", "valid_from": "2025-04-18T06:00:00Z", "valid_to": "2025-04-18T07:30:00Z", "allowlist": True, "note": "Suppresses track_origin_offshore_anomaly_score; halves mac_attribution weight", }, "geometry": {"type": "Polygon", "coordinates": [KILPILAHTI_POLYGON]}, } counts["decoy_flight_plan.geojson"] = write_geojson( OUT_STATIC / "decoy_flight_plan.geojson", [fp_feat], "s5-drone-launch-from-ship/decoy_flight_plan") # Helsinki coastal radar coverage circle (RAD-COAST-HEL-01, r=22 NM) cov_feat = { "type": "Feature", "properties": { "featureId": "radar-coverage-hel-22nm", "name": "RAD-COAST-HEL-01 coverage (22 NM)", "centre_lat": 60.1520, "centre_lon": 24.9520, "radius_nm": 22, }, "geometry": {"type": "Polygon", "coordinates": [_circle_polygon(60.1520, 24.9520, 22 * 1852.0, n=72)]}, } counts["radar_coverage_hel_22nm.geojson"] = write_geojson( OUT_STATIC / "radar_coverage_hel_22nm.geojson", [cov_feat], "s5-drone-launch-from-ship/radar_coverage_hel_22nm") return counts # --------------------------------------------------------------------------- # Historical layer — 8 weeks of ambient AIS, background MAC noise, and ~30 # prior legitimate near-shore drone flights with mixed vendor OUIs. # --------------------------------------------------------------------------- def generate_historical() -> dict[str, int]: counts: dict[str, int] = {} sensors = sensor_lookup() hist_end = datetime(2025, 4, 18, 0, 0, 0, tzinfo=UTC) hist_start = hist_end - timedelta(days=56) # 8 weeks # ---- Ambient AIS — coarse over 8 weeks ---- ais_baseline = build_ambient_ais(n_ships=80, start=hist_start, end=hist_end, seed=6001) counts["ais_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "ais_baseline.ndjson", ais_baseline, "s5-drone-launch-from-ship/ais_baseline") # ---- MAC baseline — consumer OUIs only at coastal/port sensors ---- # We slice by weeks to keep memory bounded. mac_baseline: list[MacObservation] = [] one_week = timedelta(days=7) t = hist_start bg_seed = 7000 while t < hist_end: slice_end = min(t + one_week, hist_end) bg = generate_background_macs( sensors, t, slice_end, mac_count=24, cadence_s=900.0, seed=bg_seed, ) mac_baseline.extend(bg) t = slice_end bg_seed += 1 mac_nd = [m.to_ndjson() for m in mac_baseline] counts["mac_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "mac_baseline.ndjson", mac_nd, "s5-drone-launch-from-ship/mac_baseline") counts["mac_baseline.csv"] = write_csv( OUT_HISTORICAL / "mac_baseline.csv", MAC_CSV_HEADER, [m.to_csv_row() for m in mac_baseline], "s5-drone-launch-from-ship/mac_baseline_sessions") # ---- ~30 prior legitimate near-shore drone flights ---- legit_flights = build_legitimate_drone_flights(hist_start, hist_end, n_flights=32, seed=8001) counts["legitimate_drones.ndjson"] = write_ndjson( OUT_HISTORICAL / "legitimate_drones.ndjson", legit_flights, "s5-drone-launch-from-ship/legitimate_drones") return counts def build_legitimate_drone_flights(start: datetime, end: datetime, *, n_flights: int, seed: int) -> list[dict[str, Any]]: """Synthesize ~n_flights prior legitimate near-shore drone flights with mixed vendor OUIs, filed flight plans, shore launch sites, and brief track summaries. Each NDJSON record describes one full flight.""" rng = random.Random(seed) oui_mix = [ ("Parrot", "90:3A:E6", "coastguard SAR / inspection"), ("DJI", "5C:E2:8C", "port operator inspection (Vuosaari / HaminaKotka)"), ("Autel", "00:1A:11", "infrastructure surveys"), ("Skydio", "B8:27:EB", "bridge inspections"), ("DJI Enterprise", "5C:E2:8C", "utility line inspections"), ] sites = [ ("Vuosaari", 60.2090, 25.1860), ("Kilpilahti", 60.3045, 25.5300), ("Helsinki port", 60.1530, 24.9550), ("Porkkala", 59.9800, 24.3800), ("HaminaKotka", 60.4625, 26.9450), ("Inkoo", 59.9400, 24.0050), ] flights: list[dict[str, Any]] = [] span_s = (end - start).total_seconds() for i in range(n_flights): vendor, prefix, mission = rng.choice(oui_mix) site_name, slat, slon = rng.choice(sites) t0 = start + timedelta(seconds=rng.uniform(0.0, span_s - 3600.0)) # Daytime only (06-18 UTC) t0 = t0.replace(hour=rng.randint(6, 17), minute=rng.randint(0, 59), second=0) duration_min = rng.randint(8, 35) t1 = t0 + timedelta(minutes=duration_min) suffix = ":".join(f"{rng.randint(0, 255):02X}" for _ in range(3)) mac = f"{prefix}:{suffix}" plan_id = f"FP-{vendor[:3].upper()}-{t0.strftime('%Y%m%d')}-{i:03d}" flights.append({ "flight_id": f"LEGIT-FLT-{t0.strftime('%Y%m%d')}-{i:03d}", "ts_start": iso_utc(t0), "ts_end": iso_utc(t1), "duration_min": duration_min, "mac": mac, "oui_vendor": vendor, "asset_persona": f"ASSET-DRN-LEGIT-{i:03d}", "launch_site": site_name, "launch_lat": slat, "launch_lon": slon, "shore_launched": True, "flight_plan_id": plan_id, "has_filed_flight_plan": True, "mission": mission, "track_summary": ( f"Shore launch from {site_name}; " f"{duration_min}-min inspection; " f"observed by coastal MAC sensors at typical " f"{rng.choice(['-72','-78','-83','-88'])} dBm; " f"RTB to launch site." ), "co_observed_with_ship_mac": False, "incident_score_estimate": round(rng.uniform(0.05, 0.31), 2), "alert": False, }) return flights # --------------------------------------------------------------------------- # Driver # --------------------------------------------------------------------------- def dir_size_bytes(p: Path) -> int: if not p.exists(): return 0 return sum(f.stat().st_size for f in p.rglob("*") if f.is_file()) def main() -> int: print("[S5] generating realtime layer …") rt = generate_realtime() print("[S5] generating static layer …") st = generate_static() print("[S5] generating historical layer …") hi = generate_historical() print("\n===== Scenario 05 — Drone Launch from Ship: generation summary =====") print("\n[realtime]") for k, v in rt.items(): if k.startswith("__"): continue print(f" {k:<32} rows/feats={v:>8}") print("\n[static]") for k, v in st.items(): print(f" {k:<32} features={v:>5}") print("\n[historical]") for k, v in hi.items(): print(f" {k:<32} rows={v:>8}") 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)") print("\n[scenario invariants]") print(f" AALLOTAR MMSI = 230999401") print(f" Loiter point = {LOITER_LAT}, {LOITER_LON}") print(f" Drone MAC (DJI suspect) = {DRONE_MAC}") print(f" Operator iPad MAC = {OPERATOR_IPAD_MAC}") print(f" Decoy drone MAC (Parrot) = {DECOY_DRONE_MAC}") print(f" Kilpilahti polygon ctr = lat={KILPILAHTI_CLAT:.5f}, lon={KILPILAHTI_CLON:.5f}") print(f" Dual-MAC co-obs window = {PAT_COOBS_1.isoformat()} .. {PAT_COOBS_2.isoformat()}") print(f" Co-obs MACs heard = {rt.get('__cooobs_macs', 0)} unique (expected 2)") summary = { "scenario": "s5-drone-launch-from-ship", "realtime": {k: v for k, v in rt.items() if not k.startswith("__")}, "static": st, "historical": hi, "bytes": {"realtime": rt_bytes, "static": st_bytes, "historical": hi_bytes}, "anchors": { "window_open": WINDOW_OPEN.isoformat(), "window_close": WINDOW_CLOSE.isoformat(), "loiter_start": LOITER_T0.isoformat(), "drone_takeoff": DRONE_TAKEOFF.isoformat(), "drone_land": DRONE_LAND.isoformat(), "co_obs_1": PAT_COOBS_1.isoformat(), "co_obs_2": PAT_COOBS_2.isoformat(), }, "macs": { "drone_suspect": DRONE_MAC, "operator_ipad": OPERATOR_IPAD_MAC, "drone_decoy": DECOY_DRONE_MAC, }, "polygons": { "kilpilahti_recon_centroid": {"lat": KILPILAHTI_CLAT, "lon": KILPILAHTI_CLON}, }, } (SCENARIO_DIR / "data" / "_generation_summary.json").write_text( json.dumps(summary, indent=2), encoding="utf-8") print("[done] All files written under scenarios/05-drone-launch-from-ship/data/") return 0 if __name__ == "__main__": raise SystemExit(main())