"""Scenario 09 — Porvoo Drone Incursion: data generator. Story: MV VOLKOV (MMSI 273992001, Russia-prefix, Panama-flagged, IMO watchlist) approaches Porvoo, Finland from international waters. Watchlist status escalates YELLOW → RED as the vessel crosses into Finnish territorial waters. The vessel enters the outer Porvoo archipelago, turns off AIS and goes stationary. It then launches a single drone that flies NNW inland, overflying the Kilpilahti oil refinery and continuing toward Porvoo old town. MAC sensors near the launch point detect the drone controller boot and drone device first seen. Coastal radar stations track the drone across three handover zones. The vessel re-activates AIS and exits Finnish waters heading SE. Duration: ~4.5 hours (06:00–10:30 UTC on 2025-08-14). """ from __future__ import annotations import json 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, epoch_ms, haversine_m, iso_utc, maybe_decimate_mac_ndjson, maybe_decimate_ndjson, 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, generate_background_macs, ) from generators.radar_generator import RadarTrack, emit_drone_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" random.seed(9) # --------------------------------------------------------------------------- # Time anchors # --------------------------------------------------------------------------- WINDOW_OPEN = datetime(2025, 8, 14, 4, 30, 0, tzinfo=UTC) # depart Estonia EEZ_APPROACH = datetime(2025, 8, 14, 6, 30, 0, tzinfo=UTC) NEAR_FINISH = datetime(2025, 8, 14, 7, 30, 0, tzinfo=UTC) TERRITORIAL = datetime(2025, 8, 14, 8, 15, 0, tzinfo=UTC) # crosses territorial waters VIA_EMASALO = datetime(2025, 8, 14, 8, 51, 0, tzinfo=UTC) # AIS dark at Emäsalo via point AIS_DARK = datetime(2025, 8, 14, 8, 51, 0, tzinfo=UTC) # AIS goes dark ANCHOR_ARR = datetime(2025, 8, 14, 9, 5, 0, tzinfo=UTC) # vessel stops at anchor CTRL_BOOT = datetime(2025, 8, 14, 9, 2, 0, tzinfo=UTC) # MAC: controller detected DRONE_SEEN = datetime(2025, 8, 14, 9, 5, 0, tzinfo=UTC) # MAC: drone device first seen DRONE_LAUNCH = datetime(2025, 8, 14, 9, 8, 0, tzinfo=UTC) # drone launched VESSEL_DEPART = datetime(2025, 8, 14, 9, 20, 0, tzinfo=UTC) # vessel departs anchor AIS_BACK = datetime(2025, 8, 14, 9, 34, 0, tzinfo=UTC) # AIS back on at via point T_1102 = datetime(2025, 8, 14, 11, 2, 0, tzinfo=UTC) # vessel at end point WINDOW_CLOSE = datetime(2025, 8, 14, 11, 10, 0, tzinfo=UTC) # --------------------------------------------------------------------------- # Vessel: MV VOLKOV # --------------------------------------------------------------------------- VOLKOV_MMSI = 273992001 VOLKOV_IMO = "IMO9993701" VOLKOV_NAME = "MV VOLKOV" VOLKOV_CALLSIGN = "UBVM9" VOLKOV_WAYPOINTS: list[tuple[datetime, float, float]] = [ (WINDOW_OPEN, 59.933899, 27.277518), # Narva Bay, Estonia (EEZ_APPROACH, 60.020, 26.600), # Mid-Gulf transit (NEAR_FINISH, 60.080, 26.000), # Approaching Finnish waters (TERRITORIAL, 60.140, 25.750), # Crosses territorial waters → RED (VIA_EMASALO, 60.185057, 25.611716), # Emäsalo via/stop point → AIS DARK + STOP # Vessel stationary at stop/via point throughout dark window (08:51–09:34) (AIS_BACK, 60.185057, 25.611716), # Same position → AIS back ON (T_1102, 59.906323, 25.437995), # Exit: international waters (WINDOW_CLOSE, 59.850, 25.420), # Window close ] # Dark window: 08:51 → 09:34 UTC (43 min — via point to via point) AIS_DARK_WINDOWS = [(AIS_DARK, AIS_BACK)] # --------------------------------------------------------------------------- # Sensors — S9-specific (not in catalog; generated inline) # --------------------------------------------------------------------------- S9_SENSORS: dict[str, dict[str, Any]] = { "MAC-EMÄS-BUOY-01": { "sensorId": "MAC-EMÄS-BUOY-01", "name": "Emäsalo Outer Buoy MAC", "kind": "mac", "lat": 60.181, "lon": 25.618, # 490m south of stop point (60.185, 25.612) "active_start": "08:00", "active_end": "10:00", }, "MAC-EMÄS-COAST-01": { "sensorId": "MAC-EMÄS-COAST-01", "name": "Emäsalo Approach Channel MAC", "kind": "mac", "lat": 60.194, "lon": 25.595, # 690m from drone path at T+4min "active_start": "07:00", "active_end": "11:00", }, "MAC-KILP-COAST-01": { "sensorId": "MAC-KILP-COAST-01", "name": "Kilpilahti Industrial Waterfront MAC", "kind": "mac", "lat": 60.287, "lon": 25.542, "active_start": "07:00", "active_end": "11:00", }, "RAD-EMÄS-01": { "sensorId": "RAD-EMÄS-01", "name": "Emäsalo Lighthouse Radar", "kind": "radar", "lat": 60.192, "lon": 25.640, "first_detection": "09:10", }, "RAD-KILP-01": { "sensorId": "RAD-KILP-01", "name": "Kilpilahti Industrial Radar", "kind": "radar", "lat": 60.298, "lon": 25.548, "first_detection": "09:20", }, "RAD-PORV-01": { "sensorId": "RAD-PORV-01", "name": "Porvoo Coastal Radar (Hill)", "kind": "radar", "lat": 60.388, "lon": 25.628, "first_detection": "09:28", }, } # --------------------------------------------------------------------------- # Infrastructure targets # --------------------------------------------------------------------------- S9_INFRA: list[dict[str, Any]] = [ {"featureId": "emäsalo-channel", "name": "Emäsalo Channel", "type": "maritime_access", "criticality": "medium", "lat": 60.196, "lon": 25.630}, {"featureId": "kilpilahti-refinery","name": "Kilpilahti Oil Refinery (Neste)", "type": "energy_refinery", "criticality": "critical", "lat": 60.296, "lon": 25.550}, {"featureId": "kilpilahti-lng", "name": "Kilpilahti LNG Terminal", "type": "energy_lng", "criticality": "critical", "lat": 60.292, "lon": 25.562}, {"featureId": "e18-bridge-porvoo", "name": "E18 Motorway Bridge (Porvoo)", "type": "transport", "criticality": "high", "lat": 60.346, "lon": 25.598}, {"featureId": "porvoo-substation", "name": "Porvoo Power Substation", "type": "energy_grid", "criticality": "high", "lat": 60.365, "lon": 25.615}, {"featureId": "porvoo-old-town", "name": "Porvoo Old Town / Cathedral","type": "civilian_cultural","criticality": "high", "lat": 60.393, "lon": 25.663}, ] # --------------------------------------------------------------------------- # Drone flight path — T-DRN-S9-D1 # --------------------------------------------------------------------------- DRONE_TRACK_ID = "T-DRN-S9-D1" # (time, lat, lon, alt_m, sensor_id) # Launch from stop point (60.185, 25.612), heading NNW ~344° toward Kilpilahti DRONE_WAYPOINTS_FULL: list[tuple[datetime, float, float, float, str]] = [ (DRONE_LAUNCH, 60.185, 25.612, 5, "RAD-EMÄS-01"), (DRONE_LAUNCH + timedelta(minutes=2), 60.192, 25.605, 45, "RAD-EMÄS-01"), (DRONE_LAUNCH + timedelta(minutes=4), 60.200, 25.594, 80, "RAD-EMÄS-01"), (DRONE_LAUNCH + timedelta(minutes=7), 60.216, 25.578, 95, "RAD-EMÄS-01"), (DRONE_LAUNCH + timedelta(minutes=10), 60.232, 25.567,100, "RAD-EMÄS-01"), (DRONE_LAUNCH + timedelta(minutes=12), 60.250, 25.560,100, "RAD-KILP-01"), (DRONE_LAUNCH + timedelta(minutes=14), 60.268, 25.555,102, "RAD-KILP-01"), (DRONE_LAUNCH + timedelta(minutes=17), 60.290, 25.548, 98, "RAD-KILP-01"), # Kilpilahti refinery (60.296, 25.550) (DRONE_LAUNCH + timedelta(minutes=19), 60.305, 25.552, 95, "RAD-KILP-01"), (DRONE_LAUNCH + timedelta(minutes=20), 60.318, 25.558, 92, "RAD-PORV-01"), (DRONE_LAUNCH + timedelta(minutes=22), 60.335, 25.578, 88, "RAD-PORV-01"), (DRONE_LAUNCH + timedelta(minutes=24), 60.348, 25.598, 85, "RAD-PORV-01"), # E18 bridge (60.346, 25.598) (DRONE_LAUNCH + timedelta(minutes=27), 60.360, 25.608, 80, "RAD-PORV-01"), (DRONE_LAUNCH + timedelta(minutes=30), 60.365, 25.615, 75, "RAD-PORV-01"), # substation (60.365, 25.615) (DRONE_LAUNCH + timedelta(minutes=34), 60.383, 25.638, 68, "RAD-PORV-01"), (DRONE_LAUNCH + timedelta(minutes=37), 60.393, 25.655, 60, "RAD-PORV-01"), ] # Split into per-sensor tracks (contiguous segments per sensor handover) def _split_drone_by_sensor(full: list[tuple[datetime, float, float, float, str]]) -> list[tuple[str, list[tuple[datetime, float, float]], list[tuple[datetime, float]]]]: """Return list of (sensor_id, waypoints_2d, altitudes) for each sensor segment.""" segments: list[tuple[str, list, list]] = [] cur_sensor = full[0][4] wp2d: list[tuple[datetime, float, float]] = [] alts: list[tuple[datetime, float]] = [] for t, la, lo, alt, sid in full: if sid != cur_sensor and wp2d: # overlap: carry last point into the new segment segments.append((cur_sensor, wp2d, alts)) wp2d = [wp2d[-1]] alts = [alts[-1]] cur_sensor = sid wp2d.append((t, la, lo)) alts.append((t, float(alt))) if wp2d: segments.append((cur_sensor, wp2d, alts)) return segments # --------------------------------------------------------------------------- # AIS enrichment helpers # --------------------------------------------------------------------------- def _watchlist_flag(lat: float) -> str: return "RED" if lat >= 60.000 else "YELLOW" def _in_territorial_waters(lat: float) -> bool: return lat >= 59.950 # simplified boundary def _enrich_ais(records: list[dict[str, Any]]) -> list[dict[str, Any]]: for r in records: if r.get("mmsi") == VOLKOV_MMSI: r["watchlist_flag"] = _watchlist_flag(r["lat"]) r["territorial_waters"] = _in_territorial_waters(r["lat"]) r["imo"] = VOLKOV_IMO r["name"] = VOLKOV_NAME r["callsign"] = VOLKOV_CALLSIGN r["ship_type"] = 70 r["loa_m"] = 95 r["beam_m"] = 15 r["flag"] = "PAN" return records # --------------------------------------------------------------------------- # Background AIS # --------------------------------------------------------------------------- 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): lat0 = rng.uniform(59.50, 60.35) lat1 = lat0 + rng.uniform(-0.20, 0.20) lon0 = rng.uniform(24.80, 26.20) lon1 = lon0 + rng.uniform(-0.30, 0.30) t_start = WINDOW_OPEN + timedelta(minutes=rng.uniform(0, 60)) t_end = min(WINDOW_CLOSE, t_start + timedelta(minutes=rng.uniform(60, 180))) if t_end <= t_start: continue flag_roll = rng.random() flag = "FI" if flag_roll < 0.60 else ("EE" if flag_roll < 0.80 else "OTHER") mmsi = ambient_mmsi(rng, flag) vessel_types = [30, 30, 70, 52, 31] # fishing, fishing, cargo, tug, pilot vtype = rng.choice(vessel_types) track = AisTrack( mmsi=mmsi, waypoints=[(t_start, lat0, lon0), (t_end, lat1, lon1)], cadence_s=60.0, destination="FIPRV" if rng.random() < 0.5 else "FIHEL", nav_status=0, seed=seed + i, ) recs = emit_ais(track) for r in recs: r["ship_type"] = vtype out.extend(recs) return out # --------------------------------------------------------------------------- # MAC: anomalous observations — controller boot + drone detections # --------------------------------------------------------------------------- CONTROLLER_OUI = "bc:d0:74" CONTROLLER_MAC = "bc:d0:74:4A:F3:01" DRONE_OUI = "24:0a:c4" DRONE_MAC = "24:0a:c4:BB:77:21" def emit_anomalous_macs() -> list[MacObservation]: rng = random.Random(9901) out: list[MacObservation] = [] # BUOY: controller boot at 09:02 for i in range(12): ws = CTRL_BOOT + timedelta(seconds=i * 10) we = ws + timedelta(seconds=10) out.append(MacObservation( sensor_id="MAC-EMÄS-BUOY-01", mac=CONTROLLER_MAC, session_start=ws, session_end=we, message_count=max(1, int(rng.gauss(25, 6))), avg_rssi=round(rng.uniform(-68.0, -72.0) + rng.gauss(0, 1.5), 2), manufacturer="Tello-Controller", )) # BUOY: drone OUI first seen at 09:05 for i in range(8): ws = DRONE_SEEN + timedelta(seconds=i * 12) we = ws + timedelta(seconds=12) out.append(MacObservation( sensor_id="MAC-EMÄS-BUOY-01", mac=DRONE_MAC, session_start=ws, session_end=we, message_count=max(1, int(rng.gauss(15, 4))), avg_rssi=round(rng.uniform(-70.0, -74.0) + rng.gauss(0, 1.5), 2), manufacturer="Espressif", )) # BUOY: drone signal weakening 09:08–09:15 (flying away) for i in range(7): ws = DRONE_LAUNCH + timedelta(minutes=i) we = ws + timedelta(minutes=1) rssi = -72.0 - i * 3.5 + rng.gauss(0, 1.2) out.append(MacObservation( sensor_id="MAC-EMÄS-BUOY-01", mac=DRONE_MAC, session_start=ws, session_end=we, message_count=max(1, int(rng.gauss(8, 3))), avg_rssi=round(rssi, 2), manufacturer="Espressif", )) # MAC-EMÄS-COAST-01: drone ~690m overhead at T+4min (09:12) for i in range(4): ws = DRONE_LAUNCH + timedelta(minutes=4) + timedelta(seconds=i * 30) we = ws + timedelta(seconds=30) out.append(MacObservation( sensor_id="MAC-EMÄS-COAST-01", mac=DRONE_MAC, session_start=ws, session_end=we, message_count=max(1, int(rng.gauss(18, 5))), avg_rssi=round(-60.0 + rng.gauss(0, 2.0), 2), manufacturer="Espressif", )) # MAC-KILP-COAST-01: drone ~470m overhead at T+17min (09:25) for i in range(4): ws = DRONE_LAUNCH + timedelta(minutes=16) + timedelta(seconds=i * 30) we = ws + timedelta(seconds=30) out.append(MacObservation( sensor_id="MAC-KILP-COAST-01", mac=DRONE_MAC, session_start=ws, session_end=we, message_count=max(1, int(rng.gauss(20, 5))), avg_rssi=round(-55.0 + rng.gauss(0, 2.0), 2), manufacturer="Espressif", )) return out # --------------------------------------------------------------------------- # MAC background — using S9 sensor dict directly # --------------------------------------------------------------------------- def generate_s9_background_macs(start: datetime, end: datetime, seed: int = 9042) -> list[MacObservation]: """Generate background MAC noise for the three S9 coastal/buoy sensors.""" rng = random.Random(seed) # Consumer-device OUI prefixes (common Baltic coastal background) oui_choices = [ ("Apple", "A4:C3:F0"), ("Apple", "DC:2B:2A"), ("Apple", "F8:FF:C2"), ("Samsung", "E4:A7:C5"), ("Samsung","B8:5A:73"), ("Nokia", "00:25:71"), ("Huawei", "28:31:52"), ("Intel", "8C:8D:28"), ("Intel", "D4:F5:47"), ("Ericsson", "AC:1F:6B"), ("Bosch", "00:09:02"), ("Siemens","00:0E:8C"), ("Unknown", "44:65:0D"), ("Unknown","FC:B4:E6"), ] active_sensors = { "MAC-EMÄS-BUOY-01": (datetime(2025, 8, 14, 8, 0, tzinfo=UTC), datetime(2025, 8, 14, 10, 0, tzinfo=UTC)), "MAC-EMÄS-COAST-01": (datetime(2025, 8, 14, 7, 0, tzinfo=UTC), datetime(2025, 8, 14, 11, 0, tzinfo=UTC)), "MAC-KILP-COAST-01": (datetime(2025, 8, 14, 7, 0, tzinfo=UTC), datetime(2025, 8, 14, 11, 0, tzinfo=UTC)), } # Generate ~25 unique background MACs per sensor out: list[MacObservation] = [] for sid, (s_start, s_end) in active_sensors.items(): n_bg_macs = rng.randint(20, 30) bg_macs = [] for _ in range(n_bg_macs): vendor, prefix = rng.choice(oui_choices) suffix = ":".join(f"{rng.randint(0, 255):02x}" for _ in range(3)) bg_macs.append((f"{prefix}:{suffix}", vendor)) t = s_start while t < s_end: win_end = min(s_end, t + timedelta(minutes=1)) n_active = rng.randint(20, 30) chosen = rng.sample(bg_macs, k=min(n_active, len(bg_macs))) for mac, vendor in chosen: rssi = rng.uniform(-102, -58) m_start = t if rng.random() > 0.12 else None m_end = win_end if rng.random() > 0.12 else None m_vendor = vendor if rng.random() > 0.25 else None out.append(MacObservation( sensor_id=sid, mac=mac, session_start=m_start, session_end=m_end, message_count=rng.randint(1, 50), avg_rssi=round(rssi, 2), manufacturer=m_vendor, )) t = win_end return out # --------------------------------------------------------------------------- # Drone radar — emit per-sensor-segment # --------------------------------------------------------------------------- def generate_drone_radar() -> list[dict[str, Any]]: rng = random.Random(9091) all_recs: list[dict[str, Any]] = [] segments = _split_drone_by_sensor(DRONE_WAYPOINTS_FULL) for seg_i, (sensor_id, wp2d, alts) in enumerate(segments): if len(wp2d) < 2: continue track = RadarTrack( track_id=DRONE_TRACK_ID, sensor_id=sensor_id, waypoints=wp2d, cadence_s=4.0, classification="drone_small", rcs_m2=0.025, confidence=0.80, seed=9090 + seg_i, ) recs = emit_drone_radar(track, alts) speed_choices = [s * KN_MS for s in range(28, 36)] # 28–35 kn in m/s for r in recs: r["kind"] = "airborne" r["speed_kn"] = round(r.get("speed_mps", 15.4) / KN_MS + rng.gauss(0, 0.8), 1) r.setdefault("rcs_m2", round(rng.uniform(0.015, 0.035), 3)) r["confidence"] = round(min(0.95, max(0.75, r.get("confidence", 0.80) + rng.gauss(0, 0.04))), 3) all_recs.extend(recs) return all_recs # --------------------------------------------------------------------------- # Drone snapshot GeoJSON feature # --------------------------------------------------------------------------- def drone_last_fix_feature() -> dict[str, Any]: last = DRONE_WAYPOINTS_FULL[-1] return { "type": "Feature", "geometry": {"type": "Point", "coordinates": [last[2], last[1]]}, "properties": { "track_id": DRONE_TRACK_ID, "kind": "airborne", "classification": "drone_small", "alt_m": last[3], "timestamp": iso_utc(last[0]), "ts_epoch_ms": epoch_ms(last[0]), "note": "Last radar fix — Porvoo old town", }, } # --------------------------------------------------------------------------- # Main generation functions # --------------------------------------------------------------------------- def generate_realtime() -> dict[str, int]: counts: dict[str, int] = {} # ----- AIS: MV VOLKOV ----- volkov_track = AisTrack( mmsi=VOLKOV_MMSI, waypoints=VOLKOV_WAYPOINTS, cadence_s=30.0, dark_windows=AIS_DARK_WINDOWS, destination="FIPRV", nav_status=0, seed=901, ) ship_msgs = emit_ais(volkov_track) # Force nav_status=1 (anchored) during AIS dark window approach/stop for r in ship_msgs: ts = datetime.fromtimestamp(r["ts_epoch_ms"] / 1000, tz=UTC) if AIS_DARK <= ts < AIS_BACK: r["nav_status"] = 1 ship_msgs = _enrich_ais(ship_msgs) # Background vessels (7 ships: fishing boats, cargo, ferries) ambient_msgs = build_ambient_ais(n_ships=7, seed=9000) ais_all = ship_msgs + ambient_msgs counts["ais.ndjson"] = write_ndjson( OUT_REALTIME / "ais.ndjson", ais_all, "s9-porvoo-drone/ais") snapshot_features = ais_snapshot_geojson(ais_all) snapshot_features.append(drone_last_fix_feature()) counts["ais_snapshot.geojson"] = write_geojson( OUT_REALTIME / "ais_snapshot.geojson", snapshot_features, "s9-porvoo-drone/ais_snapshot") # ----- Drone radar ----- drone_recs = generate_drone_radar() counts["drone_radar.ndjson"] = write_ndjson( OUT_REALTIME / "drone_radar.ndjson", drone_recs, "s9-porvoo-drone/drone_radar") # ----- MAC ----- anomalous = emit_anomalous_macs() background = generate_s9_background_macs(WINDOW_OPEN, WINDOW_CLOSE) mac_obs = anomalous + background mac_nd = [m.to_ndjson() for m in mac_obs] counts["mac.ndjson"] = write_ndjson( OUT_REALTIME / "mac.ndjson", mac_nd, "s9-porvoo-drone/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, "s9-porvoo-drone/mac_sessions") # ----- Decimated companions ----- AIS_DECIM_FIELDS = ["timestamp", "lat", "lon", "sog_kn", "cog_deg", "nav_status", "watchlist_flag", "territorial_waters"] DRONE_DECIM_FIELDS = ["timestamp", "lat", "lon", "alt_m", "speed_mps", "heading_deg", "rcs_m2", "classification", "kind"] 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("[S9] 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 aoi = { "type": "Feature", "properties": { "featureId": "s9-aoi", "name": "S9 Area of Interest", "note": "MV VOLKOV approach corridor + Porvoo drone flight path to old town", }, "geometry": {"type": "Polygon", "coordinates": [[ [25.40, 59.55], [27.35, 59.55], [27.35, 60.45], [25.40, 60.45], [25.40, 59.55], ]]}, } counts["area_of_interest.geojson"] = write_geojson( OUT_STATIC / "area_of_interest.geojson", [aoi], "s9-porvoo-drone/aoi") # Sensors GeoJSON — build inline from S9_SENSORS sensor_feats = [] for sid, s in S9_SENSORS.items(): sensor_feats.append({ "type": "Feature", "geometry": {"type": "Point", "coordinates": [s["lon"], s["lat"]]}, "properties": {k: v for k, v in s.items() if k not in ("lat", "lon")}, }) counts["sensors_used.geojson"] = write_geojson( OUT_STATIC / "sensors_used.geojson", sensor_feats, "s9-porvoo-drone/sensors_used") # Infrastructure GeoJSON infra_feats = [] for inf in S9_INFRA: infra_feats.append({ "type": "Feature", "geometry": {"type": "Point", "coordinates": [inf["lon"], inf["lat"]]}, "properties": {k: v for k, v in inf.items() if k not in ("lat", "lon")}, }) counts["infrastructure_used.geojson"] = write_geojson( OUT_STATIC / "infrastructure_used.geojson", infra_feats, "s9-porvoo-drone/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, 8)] for d in baseline_days: day_start = d.replace(hour=6, minute=0, second=0, microsecond=0) # Shift waypoints to historical day (no dark windows — normal transits) hist_wp = [ (day_start + timedelta(seconds=int((t - WINDOW_OPEN).total_seconds())), la, lo) for (t, la, lo) in VOLKOV_WAYPOINTS ] 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=VOLKOV_MMSI, waypoints=hist_wp, cadence_s=60.0, destination="FIPRV", seed=9900 + d.day, ) recs = emit_ais(hist_track) recs = _enrich_ais(recs) ais_baseline.extend(recs) counts["ais_baseline.ndjson"] = write_ndjson( OUT_HISTORICAL / "ais_baseline.ndjson", ais_baseline, "s9-porvoo-drone/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("[S9] generating realtime layer …") rt = generate_realtime() print("[S9] generating static layer …") st = generate_static() print("[S9] generating historical layer …") hi = generate_historical() print("\n===== Scenario 09 — Porvoo Drone Incursion: 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:<38} 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": "s9-porvoo-drone-incursion", "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/09-porvoo-drone-incursion/data/") return 0 if __name__ == "__main__": raise SystemExit(main())