"""Scenario 07 — Drone Swarm From Ship: data generator. A previously-unseen vessel (MV RASKAUS) loiters offshore 25 km south of Helsinki and releases a coordinated swarm of 5 drones toward separate Finnish coastal targets. The swarm fans out, overflies critical infrastructure (Kilpilahti, Vuosaari, Estlink corridor) and returns to the ship after ~50 min. Pipeline: 1. MV RASKAUS (MMSI 230777301) transits into the Gulf from the south, holds at (59.82°N, 24.55°E) — no NOTAMs, no port pre-arrival notice. 2. Five DJI-OUI drones (D1–D5) launch staggered 30 s apart from 09:00 UTC. Each has a distinct bearing, altitude, and target area. 3. Operator iPad MAC A4:83:E7:5C:9B:55 (on ship) is captured by coastal sensors and the patrol drone's airborne sensor MAC-AIR-DRN-01. 4. All 5 drone MACs are new to every sensor's baseline window. 5. Border Guard patrol drone RAD-DRN-PAT-01 launches from Malmi and intercepts the D1/D2 cluster; carries MAC-AIR-DRN-01 which captures BOTH D1-MAC and operator-MAC (smoking-gun co-observation). 6. After ~50 min all drones return; RASKAUS accelerates south at 15 kn. Differ from S5: S5 = single drone, ship→land. S7 = 5-drone swarm, ship→land. """ 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, 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, 5, 21, 7, 30, 0, tzinfo=UTC) WINDOW_CLOSE = datetime(2025, 5, 21, 11, 0, 0, tzinfo=UTC) LOITER_T0 = datetime(2025, 5, 21, 8, 10, 0, tzinfo=UTC) LOITER_T1 = datetime(2025, 5, 21, 10, 5, 0, tzinfo=UTC) # Swarm launch sequence (30 s stagger) LAUNCH_D1 = datetime(2025, 5, 21, 9, 0, 0, tzinfo=UTC) LAUNCH_D2 = datetime(2025, 5, 21, 9, 0, 30, tzinfo=UTC) LAUNCH_D3 = datetime(2025, 5, 21, 9, 1, 0, tzinfo=UTC) LAUNCH_D4 = datetime(2025, 5, 21, 9, 1, 30, tzinfo=UTC) LAUNCH_D5 = datetime(2025, 5, 21, 9, 2, 0, tzinfo=UTC) # All drones RTB ~09:50 RTB_START = datetime(2025, 5, 21, 9, 50, 0, tzinfo=UTC) RTB_END = datetime(2025, 5, 21, 10, 5, 0, tzinfo=UTC) # Patrol drone (RAD-DRN-PAT-01) PAT_LAUNCH = datetime(2025, 5, 21, 9, 22, 0, tzinfo=UTC) PAT_COOBS_1 = datetime(2025, 5, 21, 9, 28, 0, tzinfo=UTC) PAT_COOBS_2 = datetime(2025, 5, 21, 9, 28, 45, tzinfo=UTC) PAT_LAND = datetime(2025, 5, 21, 9, 55, 0, tzinfo=UTC) # --------------------------------------------------------------------------- # Geographic anchors # --------------------------------------------------------------------------- SHIP_LAT, SHIP_LON = 59.82, 24.55 # RASKAUS loiter point # MV RASKAUS full track RASKAUS_WP: list[tuple[datetime, float, float]] = [ (WINDOW_OPEN, 59.65, 24.55), # entry south (WINDOW_OPEN + timedelta(minutes=40), 59.82, 24.55), # loiter on-station (LOITER_T1, 59.82, 24.53), # still loitering (LOITER_T1 + timedelta(minutes=3), 59.80, 24.50), # spool up (LOITER_T1 + timedelta(minutes=25), 59.55, 24.30), # depart south ] # Drone paths (t, lat, lon, alt_m) — each drone goes to a different target then RTBs # D1: NE → Kilpilahti D1_PATH: list[tuple[datetime, float, float, float]] = [ (LAUNCH_D1, 59.82, 24.55, 0.0), (LAUNCH_D1 + timedelta(minutes=1), 59.84, 24.60, 120.0), (LAUNCH_D1 + timedelta(minutes=9), 60.10, 25.15, 122.0), (LAUNCH_D1 + timedelta(minutes=16), 60.28, 25.44, 120.0), (LAUNCH_D1 + timedelta(minutes=20), 60.32, 25.52, 115.0), # Kilpilahti orbit start (LAUNCH_D1 + timedelta(minutes=27), 60.31, 25.53, 115.0), # orbit end (RTB_START, 60.10, 25.15, 80.0), (RTB_END, 59.82, 24.55, 0.0), ] # D2: N → Vuosaari port D2_PATH: list[tuple[datetime, float, float, float]] = [ (LAUNCH_D2, 59.82, 24.55, 0.0), (LAUNCH_D2 + timedelta(minutes=1), 59.84, 24.56, 110.0), (LAUNCH_D2 + timedelta(minutes=10), 60.10, 24.80, 112.0), (LAUNCH_D2 + timedelta(minutes=17), 60.21, 25.04, 110.0), # Vuosaari (LAUNCH_D2 + timedelta(minutes=21), 60.21, 25.05, 105.0), (RTB_START + timedelta(seconds=120), 60.05, 24.80, 60.0), (RTB_END, 59.82, 24.55, 0.0), ] # D3: NW → Inkoo / Estlink corridor D3_PATH: list[tuple[datetime, float, float, float]] = [ (LAUNCH_D3, 59.82, 24.55, 0.0), (LAUNCH_D3 + timedelta(minutes=1), 59.84, 24.50, 110.0), (LAUNCH_D3 + timedelta(minutes=10), 59.95, 24.20, 112.0), (LAUNCH_D3 + timedelta(minutes=18), 59.97, 23.82, 110.0), # Estlink-1 hover (LAUNCH_D3 + timedelta(minutes=22), 59.97, 23.82, 108.0), # 4-min dwell (RTB_START, 59.92, 24.10, 70.0), (RTB_END, 59.82, 24.55, 0.0), ] # D4: N/NNW → Helsinki South Harbour D4_PATH: list[tuple[datetime, float, float, float]] = [ (LAUNCH_D4, 59.82, 24.55, 0.0), (LAUNCH_D4 + timedelta(minutes=1), 59.84, 24.55, 110.0), (LAUNCH_D4 + timedelta(minutes=14), 60.15, 24.95, 108.0), # Helsinki harbour approach (LAUNCH_D4 + timedelta(minutes=18), 60.15, 24.97, 105.0), (RTB_START + timedelta(seconds=60), 60.00, 24.75, 60.0), (RTB_END, 59.82, 24.55, 0.0), ] # D5: NE/E → Porvoo coast approach (eastern flank) D5_PATH: list[tuple[datetime, float, float, float]] = [ (LAUNCH_D5, 59.82, 24.55, 0.0), (LAUNCH_D5 + timedelta(minutes=1), 59.84, 24.62, 115.0), (LAUNCH_D5 + timedelta(minutes=10), 60.05, 25.20, 118.0), (LAUNCH_D5 + timedelta(minutes=17), 60.22, 25.60, 115.0), # Porvoo coast (LAUNCH_D5 + timedelta(minutes=21), 60.22, 25.62, 112.0), (RTB_START + timedelta(seconds=180), 60.05, 25.10, 60.0), (RTB_END, 59.82, 24.55, 0.0), ] ALL_DRONE_PATHS: list[tuple[str, list[tuple[datetime, float, float, float]]]] = [ ("D1", D1_PATH), ("D2", D2_PATH), ("D3", D3_PATH), ("D4", D4_PATH), ("D5", D5_PATH), ] # Patrol drone PATROL_PATH: list[tuple[datetime, float, float, float]] = [ (PAT_LAUNCH, 60.25, 25.04, 0.0), (PAT_LAUNCH + timedelta(minutes=4), 60.22, 25.15, 240.0), (PAT_LAUNCH + timedelta(minutes=10), 60.10, 24.95, 240.0), (PAT_COOBS_1, 59.93, 24.62, 220.0), (PAT_COOBS_2, 59.94, 24.64, 220.0), (PAT_LAND, 60.25, 25.04, 0.0), ] # --------------------------------------------------------------------------- # MAC IDs # --------------------------------------------------------------------------- DRONE_MACS = { "D1": ("5C:E2:8C:DD:EE:11", "DJI"), "D2": ("5C:E2:8C:DD:EE:12", "DJI"), "D3": ("5C:E2:8C:DD:EE:13", "DJI"), "D4": ("5C:E2:8C:DD:EE:14", "DJI"), "D5": ("5C:E2:8C:DD:EE:15", "DJI"), } OPERATOR_MAC = "A4:83:E7:5C:9B:55" # iPad on ship OPERATOR_VENDOR = "Apple" RASKAUS_CREW_MACS = [ ("38:F9:D3:AB:CD:01", "Samsung", "Master"), ("38:F9:D3:AB:CD:02", "Samsung", "Chief Officer"), ("A4:83:E7:5C:9B:55", "Apple", "Operator"), ] # Coastal sensors with LOS to drone overflight tracks COASTAL_DRONE_SENSORS = [ "MAC-PRV-COAST-02", "MAC-PRV-COAST-01", "MAC-HEL-COAST-01", "MAC-INK-COAST-01", "MAC-PRK-COAST-01", ] SENSORS_USED_IDS = { "MAC-PRV-COAST-02", "MAC-PRV-COAST-01", "MAC-HEL-COAST-01", "MAC-INK-COAST-01", "MAC-PRK-COAST-01", "MAC-AIR-DRN-01", "MAC-AIR-PLN-01", "RAD-PLN-01", "RAD-DRN-PAT-01", } INFRA_USED_IDS = { "estlink-1", "estlink-1-buffer", "site-kilpilahti", "port-kilpilahti", "site-vuosaari", "port-helsinki", "shipping-lane-eb", "shipping-lane-wb", } # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- 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 _track_pos_3d( track: list[tuple[datetime, float, float, float]], t: datetime ) -> tuple[float, float, float]: if t <= track[0][0]: return track[0][1], track[0][2], track[0][3] if t >= track[-1][0]: return track[-1][1], track[-1][2], track[-1][3] 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 lat = a[1] + f * (b[1] - a[1]) lon = a[2] + f * (b[2] - a[2]) alt = a[3] + f * (b[3] - a[3]) return lat, lon, alt return track[-1][1], track[-1][2], track[-1][3] def emit_drone_mac_at_coastal_sensors( path_3d: list[tuple[datetime, float, float, float]], mac: str, manufacturer: str, sensors: dict[str, dict[str, Any]], sensor_ids: list[str], *, 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 = 0, ) -> list[MacObservation]: """Walk the drone path in session_window_s slices and emit MAC observations at each listed coastal sensor where RSSI exceeds the threshold.""" rng = random.Random(seed) out: list[MacObservation] = [] t_start, t_end = path_3d[0][0], path_3d[-1][0] t = t_start path_2d = [(ts, lat, lon) for (ts, lat, lon, _) in path_3d] while t < t_end: seg_end = min(t_end, t + timedelta(seconds=session_window_s)) tm = t + (seg_end - t) / 2 lat, lon = _track_pos_2d(path_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 if rssi < -82.0 and rng.random() < 0.5: continue out.append(MacObservation( sensor_id=sid, mac=mac, session_start=t, session_end=seg_end, message_count=max(1, int(rng.gauss(6, 2))), avg_rssi=rssi, manufacturer=manufacturer, version="1.4", status="ACTIVE" if rssi > -80.0 else "FRINGE", )) t = seg_end return out def emit_airborne_sensor_capture( patrol_path: list[tuple[datetime, float, float, float]], drone_path: list[tuple[datetime, float, float, float]], target_mac: str, target_vendor: 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 = 0, ) -> list[MacObservation]: """Airborne patrol sensor captures the target MAC. Uses 3D distance.""" rng = random.Random(seed) out: list[MacObservation] = [] t_start, t_end = active_window t = t_start p_path_2d = [(ts, lat, lon) for ts, lat, lon, _ in patrol_path] d_path_2d = [(ts, lat, lon) for ts, lat, lon, _ in drone_path] while t < t_end: seg_end = min(t_end, t + timedelta(seconds=session_window_s)) tm = t + (seg_end - t) / 2 slat, slon, salt = _track_pos_3d(patrol_path, tm) dlat, dlon, dalt = _track_pos_3d(drone_path, tm) horiz = haversine_m(slat, slon, dlat, dlon) vert = abs(salt - dalt) d3 = math.sqrt(horiz ** 2 + vert ** 2) rssi = tx_dbm - 10.0 * n * math.log10(max(d3, 1.0)) + rng.gauss(0.0, noise_db) if rssi >= rssi_threshold: out.append(MacObservation( sensor_id="MAC-AIR-DRN-01", mac=target_mac, session_start=t, session_end=seg_end, message_count=max(1, int(rng.gauss(4, 1))), avg_rssi=rssi, manufacturer=target_vendor, version="1.4", status="ACTIVE" if rssi > -80.0 else "FRINGE", )) t = seg_end return out def emit_drone_radar_track( drone_id: str, path_3d: list[tuple[datetime, float, float, float]], rng: random.Random, ) -> list[dict[str, Any]]: """Emit radar records for one drone along its 3D path.""" records = [] t = path_3d[0][0] t_end = path_3d[-1][0] cadence = timedelta(seconds=4) while t <= t_end: lat, lon, alt = _track_pos_3d(path_3d, t) if alt > 5.0: # airborne only records.append({ "__meta__": "synthetic", "timestamp": iso_utc(t), "ts_epoch_ms": int(t.timestamp() * 1000), "track_id": f"T-DRN-S7-{drone_id}", "sensor_id": "RAD-PLN-01", "lat": round(lat + rng.gauss(0, 0.0001), 6), "lon": round(lon + rng.gauss(0, 0.0001), 6), "alt_m": round(alt + rng.gauss(0, 2.0), 1), "speed_mps": round(rng.gauss(18.0, 1.5), 1), "heading_deg": round(rng.uniform(0, 360), 1), "rcs_m2": round(rng.gauss(0.02, 0.005), 4), "classification": "drone_small", "confidence": round(rng.gauss(0.82, 0.05), 3), "kind": "airborne", }) t += cadence return records # --------------------------------------------------------------------------- # Ambient AIS helpers (same pattern as other scenarios) # --------------------------------------------------------------------------- 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.65, 60.10) lat1 = lat0 + rng.uniform(-0.08, 0.08) lon0, lon1 = (22.5, 27.5) if eastbound else (27.5, 22.5) t_start = WINDOW_OPEN + timedelta(minutes=rng.uniform(0, 150)) t_end = min(WINDOW_CLOSE, t_start + timedelta(minutes=rng.uniform(60, 200))) if t_end <= t_start: continue flag_roll = rng.random() flag = "FI" if flag_roll < 0.7 else ("EE" if flag_roll < 0.9 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 # --------------------------------------------------------------------------- # Main generators # --------------------------------------------------------------------------- def generate_realtime() -> dict[str, int]: sensors = sensor_lookup() counts: dict[str, int] = {} # ---- RASKAUS AIS ---- raskaus_track = AisTrack( mmsi=230777301, waypoints=RASKAUS_WP, cadence_s=4.0, destination="FIHAN", nav_status=0, seed=601, ) raskaus_msgs = emit_ais(raskaus_track) ambient_msgs = build_ambient_ais(n_ships=1200, seed=602) ais_all = raskaus_msgs + ambient_msgs counts["ais.ndjson"] = write_ndjson( OUT_REALTIME / "ais.ndjson", ais_all, "s7-drone-swarm-from-ship/ais") snapshot_features = ais_snapshot_geojson(ais_all) counts["ais_snapshot.geojson"] = write_geojson( OUT_REALTIME / "ais_snapshot.geojson", snapshot_features, "s7-drone-swarm-from-ship/ais_snapshot") # ---- Drone radar (all 5 drones) ---- rng = random.Random(610) drone_radar_all: list[dict[str, Any]] = [] for drone_id, path_3d in ALL_DRONE_PATHS: drone_radar_all.extend(emit_drone_radar_track(drone_id, path_3d, rng)) # Patrol drone radar track for r in emit_drone_radar_track("PAT", PATROL_PATH, rng): r["track_id"] = "T-DRN-PAT-S7" r["sensor_id"] = "RAD-DRN-PAT-01" r["classification"] = "drone_large" r["rcs_m2"] = round(rng.gauss(0.35, 0.04), 4) drone_radar_all.append(r) counts["drone_radar.ndjson"] = write_ndjson( OUT_REALTIME / "drone_radar.ndjson", drone_radar_all, "s7-drone-swarm-from-ship/drone_radar") # ---- MAC observations ---- mac_obs: list[MacObservation] = [] # Operator / crew MACs on RASKAUS (visible at coastal sensors during loiter) raskaus_wp_2d = [(t, lat, lon) for t, lat, lon in RASKAUS_WP] for mac, vendor, _ in RASKAUS_CREW_MACS: em = MovingMacEmitter( mac=mac, manufacturer=vendor, waypoints=raskaus_wp_2d, active_windows=[(WINDOW_OPEN, WINDOW_CLOSE)], seed=620, ) mac_obs.extend(simulate_moving_mac(em, sensors, session_window_s=120.0)) # All 5 drone MACs at coastal sensors for di, (drone_id, path_3d) in enumerate(ALL_DRONE_PATHS): mac, vendor = DRONE_MACS[drone_id] obs = emit_drone_mac_at_coastal_sensors( path_3d, mac, vendor, sensors, COASTAL_DRONE_SENSORS, seed=630 + di, ) mac_obs.extend(obs) # Patrol airborne sensor captures D1 MAC (smoking gun) + operator MAC mac_obs.extend(emit_airborne_sensor_capture( PATROL_PATH, D1_PATH, DRONE_MACS["D1"][0], DRONE_MACS["D1"][1], active_window=(PAT_LAUNCH, PAT_LAND), seed=640, )) # Operator stays on ship at sea level — convert RASKAUS_WP to 4-tuple path raskaus_path_3d = [(t, lat, lon, 0.0) for (t, lat, lon) in RASKAUS_WP] mac_obs.extend(emit_airborne_sensor_capture( PATROL_PATH, raskaus_path_3d, OPERATOR_MAC, OPERATOR_VENDOR, active_window=(PAT_COOBS_1, PAT_LAND), seed=641, )) # Background MAC noise bg = generate_background_macs(sensors, WINDOW_OPEN, WINDOW_CLOSE, mac_count=120, 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, "s7-drone-swarm-from-ship/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, "s7-drone-swarm-from-ship/mac_sessions") # Decimate large files AIS_FIELDS = ["timestamp", "lat", "lon", "sog_kn", "cog_deg", "nav_status"] DRONE_FIELDS = ["timestamp", "lat", "lon", "alt_m", "speed_mps", "rcs_m2", "classification", "kind", "track_id"] for path, kw in [ (OUT_REALTIME / "ais.ndjson", {"key_field": "mmsi", "ts_field": "ts_epoch_ms", "project_fields": AIS_FIELDS}), (OUT_REALTIME / "drone_radar.ndjson", {"key_field": "track_id", "ts_field": "ts_epoch_ms", "project_fields": DRONE_FIELDS}), ]: rep = maybe_decimate_ndjson(path, **kw) if rep: counts[Path(rep["decimated"]).name] = rep["rows"] + 1 mac_rep = maybe_decimate_mac_ndjson(OUT_REALTIME / "mac.ndjson") if mac_rep: counts[Path(mac_rep["decimated"]).name] = mac_rep["rows"] + 1 return counts def generate_static() -> dict[str, int]: counts: dict[str, int] = {} aoi = { "type": "Feature", "properties": {"featureId": "s7-aoi", "name": "S7 — Drone Swarm Area of Interest"}, "geometry": {"type": "Polygon", "coordinates": [[ [23.0, 59.55], [26.5, 59.55], [26.5, 60.45], [23.0, 60.45], [23.0, 59.55], ]]}, } counts["area_of_interest.geojson"] = write_geojson( OUT_STATIC / "area_of_interest.geojson", [aoi], "s7-drone-swarm-from-ship/aoi") sensors_fc = load_infrastructure().__class__ # just need the load pattern from generators.common import load_sensors 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, "s7-drone-swarm-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, "s7-drone-swarm-from-ship/infra_used") return counts def generate_historical() -> dict[str, int]: """7 days of ambient AIS; no swarm precedent — the baseline is clean.""" counts: dict[str, int] = {} sensors = sensor_lookup() ais_hist: list[dict[str, Any]] = [] mac_hist: list[MacObservation] = [] for day_offset in range(1, 8): day_start = WINDOW_OPEN - timedelta(days=day_offset) ais_hist.extend(build_ambient_ais(n_ships=400, seed=9000 + day_offset)) bg = generate_background_macs( sensors, day_start, day_start + timedelta(hours=6), mac_count=60, cadence_s=300.0, seed=10000 + day_offset, ) mac_hist.extend(bg) counts["ais_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "ais_baseline.ndjson", ais_hist, "s7-drone-swarm-from-ship/ais_baseline") mac_nd = [m.to_ndjson() for m in mac_hist] counts["mac_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "mac_baseline.ndjson", mac_nd, "s7-drone-swarm-from-ship/mac_baseline") mac_rows = [m.to_csv_row() for m in mac_hist] counts["mac_baseline.csv"] = write_csv( OUT_HISTORICAL / "mac_baseline.csv", MAC_CSV_HEADER, mac_rows, "s7-drone-swarm-from-ship/mac_baseline_csv") return counts def main() -> int: print("[S6] generating realtime layer …") rt = generate_realtime() print("[S6] generating static layer …") st = generate_static() print("[S6] generating historical layer …") hi = generate_historical() print("\n===== Scenario 07 — Drone Swarm From Ship: generation summary =====") print("\n[realtime]") for k, v in rt.items(): print(f" {k:<32} rows={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}") summary = {"scenario": "s7-drone-swarm-from-ship", "realtime": rt, "static": st, "historical": hi} (SCENARIO_DIR / "data" / "_generation_summary.json").write_text( json.dumps(summary, indent=2), encoding="utf-8") print("[done]") return 0 if __name__ == "__main__": raise SystemExit(main())