+
+[](https://www.python.org/downloads/)
+
+
+A suite of tools for aircraft trajectory and scenario generation, supporting both real ADS-B data processing and artificial track generation.
+
+[Installation](#-installation) • [Quick Start](#-quick-start)
+
+
+
+## 🌟 Features
+
+
+
+| Category | Features |
+|----------|----------|
+| 📊 **OpenADSB Track Generator** | • ADS‑B JSON to CSV conversion
• Geographic filtering
• Time range filtering
• Relative ownship path filtering
• Trajectory JSON conversion
• Simulation scenario creation |
+| 🎮 **Artificial Track Generator** | • Random track generation
• Circular/elliptical patterns
• Flyby/square/rectangle patterns
• Zigzag and spiral patterns
• Ownship-relative patterns
• KML data conversion |
+| 🗺️ **Google Earth Integration** | • KML to CSV conversion
• Real-world track integration
• Custom track creation |
+| 📈 **Visualization & Conversion** | • Interactive Folium maps
• Combined KML generation
• Real-time visualization |
+| ⚙️ **Scenario Generation** | • JSON scenario creation
• Actor configuration
• Sensor setup
• FMU model integration |
+| 📝 **Scenario Reports** | • Markdown report generation
• Trajectory visualization |
+| 🛠️ **Utilities** | • LLA/NED coordinate conversion
• Distance & bearing calculations
• Interactive track visualization |
+
+
+
+## 🚀 Quick Start
+
+### Installation
+
+```bash
+# Install from PyPI (Recommended)
+pip install aerosim-utils
+
+# Or install from source
+cd aerosim-utils
+pip install -e .
+```
+
+### Basic Usage
+
+
+
+[](https://opensky-network.org/)
+
+
+
+The OpenSky Network provides open access to real‑time and historical ADS‑B data:
+
+- Registration: Sign up for a free account
+- Data Access: Use their REST API for real‑time or historical data
+- Usage: Save JSON data in the `inputs/` folder
+
+**Resources:**
+- [OpenSky Network Documentation](https://opensky-network.org/)
+- [MIT LL - em-download-opensky](https://github.com/mit-ll/em-download-opensky)
+
+### ADS-B Exchange
+
+
+
+[](https://www.adsbexchange.com/)
+
+
+
+ADSBExchange offers ADS‑B data services:
+
+- Free Feed: Available for non‑commercial projects
+- API Access: Follow their documentation and guidelines
+- Integration: Store data in `inputs/` directory
+
+## 📝 Creating Tracks in Google Earth
+
+
+
+| Workflow | Description | Command |
+|----------|-------------|---------|
+| OpenADSB | Process real ADS‑B data | `aerosim workflow openadsb` |
+| Artificial | Generate artificial tracks | `aerosim workflow artificial` |
+| KML | Process Google Earth tracks | `aerosim workflow kml` |
+
+
+
+
+
diff --git a/aerosim/src/aerosim/utils/__init__.py b/aerosim/src/aerosim/utils/__init__.py
index 632cb0d..a987ee2 100644
--- a/aerosim/src/aerosim/utils/__init__.py
+++ b/aerosim/src/aerosim/utils/__init__.py
@@ -1,9 +1,108 @@
-"""
-Utilities module for AeroSim.
-
-This module provides common utility functions for the AeroSim package.
-"""
-
-from .helpers import clamp, normalize_heading_deg, distance_m_bearing_deg
-
-__all__ = ["clamp", "normalize_heading_deg", "distance_m_bearing_deg"]
+"""
+Utils - A suite of tools for aircraft trajectory and scenario generation.
+
+This package provides tools for:
+- Converting between different trajectory formats (JSON, CSV, KML)
+- Generating artificial aircraft tracks
+- Processing real ADS-B data
+- Creating simulation scenarios
+- Visualizing trajectories
+- Generating reports and plots
+
+The package includes a configuration system that can be managed through the CLI
+or programmatically. Default settings can be overridden through a config file
+or environment variables.
+"""
+
+# utils/__init__.py
+
+from .helpers import clamp, normalize_heading_deg, distance_m_bearing_deg
+from .artificial_tracks import ArtificialTrackGenerator
+from .conversion import (
+ convert_json_to_csv,
+ filter_tracks,
+ convert_tracks_to_json,
+ process_csv
+)
+from .scenario_generator import generate_scenario_json
+from .scenario_report import (
+ generate_markdown_report,
+ save_markdown_report,
+ plot_trajectories
+)
+from .tracks_from_map import (
+ extract_track_id,
+ parse_kml,
+ generate_csv_from_tracks
+)
+from .utils import (
+ geodetic_to_ecef,
+ ecef_to_geodetic,
+ lla_to_ned,
+ ned_to_lla,
+ haversine_distance,
+ euclidean_distance_lla,
+ bearing_between_points
+)
+from .visualization import visualize_folder, plot_tracks
+from .process_workflows import (
+ process_openadsb_workflow,
+ process_artificial_workflow,
+ process_kml_workflow
+)
+from .config import Config
+
+__all__ = [
+ # Helper functions
+ "clamp",
+ "normalize_heading_deg",
+ "distance_m_bearing_deg",
+
+ # Track generation
+ "ArtificialTrackGenerator",
+
+ # Conversion functions
+ "convert_json_to_csv",
+ "filter_tracks",
+ "convert_tracks_to_json",
+ "process_csv",
+
+ # Scenario generation
+ "generate_scenario_json"
+
+ # Reporting
+ "generate_markdown_report",
+ "save_markdown_report",
+ "plot_trajectories"
+
+ # Map track processing
+ "extract_track_id",
+ "parse_kml",
+ "generate_csv_from_tracks",
+
+ # Coordinate conversion
+ "geodetic_to_ecef",
+ "ecef_to_geodetic",
+ "lla_to_ned",
+ "ned_to_lla",
+ "haversine_distance",
+ "euclidean_distance_lla",
+ "bearing_between_points",
+
+ # Visualization
+ "visualize_folder",
+ "plot_tracks",
+
+ # Workflows
+ "process_openadsb_workflow",
+ "process_artificial_workflow",
+ "process_kml_workflow",
+
+ # Configuration
+ "Config"
+]
+
+__version__ = '0.1.0'
+__author__ = 'Aerosim'
+__license__ = 'MIT'
+__description__ = 'A suite of tools for aircraft trajectory and scenario generation'
diff --git a/aerosim/src/aerosim/utils/artificial_tracks.py b/aerosim/src/aerosim/utils/artificial_tracks.py
new file mode 100644
index 0000000..91b94eb
--- /dev/null
+++ b/aerosim/src/aerosim/utils/artificial_tracks.py
@@ -0,0 +1,546 @@
+import math
+import csv
+import json
+import random
+from datetime import datetime, timedelta, timezone
+
+# --- Helper Functions ---
+
+def haversine(lon1, lat1, lon2, lat2):
+ """Calculate the haversine distance between two lat/lon points in meters."""
+ R = 6371000 # Earth radius in meters
+ phi1 = math.radians(lat1)
+ phi2 = math.radians(lat2)
+ d_phi = math.radians(lat2 - lat1)
+ d_lambda = math.radians(lon2 - lon1)
+ a = math.sin(d_phi / 2)**2 + math.cos(phi1) * math.cos(phi2) * math.sin(d_lambda / 2)**2
+ c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
+ return R * c
+
+def load_ownship_trajectory(file_path, file_format="json"):
+ """
+ Load an ownship trajectory from a file.
+
+ For JSON, expect a list of dictionaries with keys: "time", "lat", "lon", "alt".
+ For CSV, assume the first three columns are lat, lon, alt and optionally a fourth column for time.
+ If time is missing, time values are assigned sequentially (e.g., 0, 1, 2, … seconds).
+ """
+ if file_format.lower() == "json":
+ with open(file_path, "r", encoding="utf-8") as f:
+ data = json.load(f)
+ # Normalize keys
+ for pt in data:
+ if "lat" not in pt and "latitude" in pt:
+ pt["lat"] = pt["latitude"]
+ if "lon" not in pt and "longitude" in pt:
+ pt["lon"] = pt["longitude"]
+ if "alt" not in pt and "altitude" in pt:
+ pt["alt"] = pt["altitude"]
+ return data
+ elif file_format.lower() == "csv":
+ trajectory = []
+ with open(file_path, newline='', encoding="utf-8") as csvfile:
+ reader = csv.reader(csvfile)
+ for i, row in enumerate(reader):
+ try:
+ lat = float(row[0])
+ lon = float(row[1])
+ alt = float(row[2])
+ except Exception:
+ continue
+ time_val = float(row[3]) if len(row) > 3 else float(i)
+ trajectory.append({"time": time_val, "lat": lat, "lon": lon, "alt": alt})
+ return trajectory
+ else:
+ raise ValueError("Unsupported ownship format. Use 'json' or 'csv'.")
+
+# --- Base and Generator Classes ---
+
+class BaseTrackGenerator:
+ def __init__(self, center_lat, center_lon, center_alt,
+ min_alt=None, max_alt=None,
+ track_id=10000, source_id="ARTIFICIAL",
+ track_type="Surveillance", altitude_reference="MSL"):
+ self.center_lat = center_lat
+ self.center_lon = center_lon
+ self.center_alt = center_alt
+ self.min_alt = min_alt if min_alt is not None else center_alt
+ self.max_alt = max_alt if max_alt is not None else center_alt
+ self.track_id = track_id # This will be a 5-digit number (e.g., 10000)
+ self.source_id = source_id
+ self.track_type = track_type
+ self.altitude_reference = altitude_reference
+ self.source_track_id = f"{source_id}::{track_id}"
+ self.start_time = datetime.now(timezone.utc)
+
+ def _create_track_point(self, lon, lat, alt, time_delta):
+ """Creates a dictionary representing a track point."""
+ timestamp = self.start_time + timedelta(seconds=time_delta)
+ hex_icao = format(random.randint(0x100000, 0xFFFFFF), 'x')
+ data_field = f"Hex-ICAO:{hex_icao}|Alt-Baro:{int(alt)}"
+ distance = haversine(self.center_lon, self.center_lat, lon, lat)
+ # Format trackId as a 5-digit string
+ return {
+ "eventName": "track.appended",
+ "trackId": f"{self.track_id:05d}",
+ "sourceId": self.source_id,
+ "trackType": self.track_type,
+ "timestamp": timestamp.isoformat() + "Z",
+ "data": data_field,
+ "altitudeReference": self.altitude_reference,
+ "sourceTrackId": self.source_track_id,
+ "longitude": lon,
+ "latitude": lat,
+ "altitude": alt,
+ "aglAltitude": alt * 0.35, # Example AGL value
+ "mslAltitude": alt,
+ "wgs84Altitude": alt - 50, # Example offset
+ "distance_m": distance
+ }
+
+ def _get_altitude(self):
+ """Randomly choose an altitude between min_alt and max_alt."""
+ return random.uniform(self.min_alt, self.max_alt)
+
+ def generate(self, num_points=10, interval_seconds=10):
+ """
+ Must be implemented by subclasses to generate track points.
+ Returns a list of track point dictionaries.
+ """
+ raise NotImplementedError("Subclasses must implement this method.")
+
+# --- Specific Maneuver Generators ---
+
+class RandomTrackGenerator(BaseTrackGenerator):
+ def generate(self, num_points=10, interval_seconds=10):
+ points = []
+ for i in range(num_points):
+ delta_lat = random.uniform(-0.01, 0.01)
+ delta_lon = random.uniform(-0.01, 0.01)
+ lat = self.center_lat + delta_lat
+ lon = self.center_lon + delta_lon
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+class CircularTrackGenerator(BaseTrackGenerator):
+ def __init__(self, center_lat, center_lon, center_alt, radius=0.005, **kwargs):
+ super().__init__(center_lat, center_lon, center_alt, **kwargs)
+ self.radius = radius
+
+ def generate(self, num_points=20, interval_seconds=10):
+ points = []
+ for i in range(num_points):
+ angle = (2 * math.pi / num_points) * i
+ delta_lat = self.radius * math.cos(angle)
+ delta_lon = self.radius * math.sin(angle)
+ lat = self.center_lat + delta_lat
+ lon = self.center_lon + delta_lon
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+class EllipticalTrackGenerator(BaseTrackGenerator):
+ def __init__(self, center_lat, center_lon, center_alt, radius_x=0.01, radius_y=0.005, **kwargs):
+ super().__init__(center_lat, center_lon, center_alt, **kwargs)
+ self.radius_x = radius_x
+ self.radius_y = radius_y
+
+ def generate(self, num_points=20, interval_seconds=10):
+ points = []
+ for i in range(num_points):
+ angle = (2 * math.pi / num_points) * i
+ delta_lat = self.radius_y * math.sin(angle)
+ delta_lon = self.radius_x * math.cos(angle)
+ lat = self.center_lat + delta_lat
+ lon = self.center_lon + delta_lon
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+class FlybyTrackGenerator(BaseTrackGenerator):
+ def __init__(self, center_lat, center_lon, center_alt,
+ approach_distance=0.02, exit_distance=0.02, bearing=45, **kwargs):
+ super().__init__(center_lat, center_lon, center_alt, **kwargs)
+ self.approach_distance = approach_distance
+ self.exit_distance = exit_distance
+ self.bearing = math.radians(bearing)
+
+ def generate(self, num_points=10, interval_seconds=10):
+ points = []
+ total_distance = self.approach_distance + self.exit_distance
+ for i in range(num_points):
+ frac = i / (num_points - 1)
+ offset = -self.approach_distance + total_distance * frac
+ delta_lat = offset * math.cos(self.bearing)
+ delta_lon = offset * math.sin(self.bearing)
+ lat = self.center_lat + delta_lat
+ lon = self.center_lon + delta_lon
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+class SquareTrackGenerator(BaseTrackGenerator):
+ def __init__(self, center_lat, center_lon, center_alt, side_length=0.02, **kwargs):
+ super().__init__(center_lat, center_lon, center_alt, **kwargs)
+ self.side_length = side_length
+
+ def generate(self, num_points=12, interval_seconds=10):
+ half = self.side_length / 2
+ corners = [
+ (self.center_lat - half, self.center_lon - half),
+ (self.center_lat - half, self.center_lon + half),
+ (self.center_lat + half, self.center_lon + half),
+ (self.center_lat + half, self.center_lon - half),
+ ]
+ path = []
+ for i in range(len(corners)):
+ path.append(corners[i])
+ next_corner = corners[(i + 1) % len(corners)]
+ mid_lat = (corners[i][0] + next_corner[0]) / 2
+ mid_lon = (corners[i][1] + next_corner[1]) / 2
+ path.append((mid_lat, mid_lon))
+ step = max(1, len(path) // num_points)
+ points = []
+ for i, (lat, lon) in enumerate(path[::step]):
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+class RectangleTrackGenerator(BaseTrackGenerator):
+ def __init__(self, center_lat, center_lon, center_alt, width=0.03, height=0.01, **kwargs):
+ super().__init__(center_lat, center_lon, center_alt, **kwargs)
+ self.width = width
+ self.height = height
+
+ def generate(self, num_points=12, interval_seconds=10):
+ half_w = self.width / 2
+ half_h = self.height / 2
+ corners = [
+ (self.center_lat - half_h, self.center_lon - half_w),
+ (self.center_lat - half_h, self.center_lon + half_w),
+ (self.center_lat + half_h, self.center_lon + half_w),
+ (self.center_lat + half_h, self.center_lon - half_w),
+ ]
+ path = []
+ for i in range(len(corners)):
+ path.append(corners[i])
+ next_corner = corners[(i + 1) % len(corners)]
+ mid_lat = (corners[i][0] + next_corner[0]) / 2
+ mid_lon = (corners[i][1] + next_corner[1]) / 2
+ path.append((mid_lat, mid_lon))
+ step = max(1, len(path) // num_points)
+ points = []
+ for i, (lat, lon) in enumerate(path[::step]):
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+
+class ZigzagTrackGenerator(BaseTrackGenerator):
+ def __init__(self, center_lat, center_lon, center_alt, direction=0, amplitude=0.005, frequency=2, **kwargs):
+ """
+ direction: main heading in degrees.
+ amplitude: lateral offset in degrees.
+ frequency: number of zigzags over the course of the track.
+ """
+ super().__init__(center_lat, center_lon, center_alt, **kwargs)
+ self.direction = math.radians(direction)
+ self.amplitude = amplitude
+ self.frequency = frequency
+
+ def generate(self, num_points=10, interval_seconds=10):
+ points = []
+ total_distance = 0.01 # total progress in degrees along main heading
+ for i in range(num_points):
+ fraction = i / (num_points - 1) if num_points > 1 else 0
+ main_offset = total_distance * fraction
+ lateral_offset = self.amplitude * math.sin(2 * math.pi * self.frequency * fraction)
+ delta_lat_main = main_offset * math.cos(self.direction)
+ delta_lon_main = main_offset * math.sin(self.direction)
+ delta_lat_perp = lateral_offset * math.cos(self.direction + math.pi / 2)
+ delta_lon_perp = lateral_offset * math.sin(self.direction + math.pi / 2)
+ lat = self.center_lat + delta_lat_main + delta_lat_perp
+ lon = self.center_lon + delta_lon_main + delta_lon_perp
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+class SpiralTrackGenerator(BaseTrackGenerator):
+ def __init__(self, center_lat, center_lon, center_alt, initial_radius=0.002, radius_increment=0.001, rotations=3, **kwargs):
+ """
+ initial_radius: starting radius in degrees.
+ radius_increment: increase in radius per point (in degrees).
+ rotations: total number of rotations over the track.
+ """
+ super().__init__(center_lat, center_lon, center_alt, **kwargs)
+ self.initial_radius = initial_radius
+ self.radius_increment = radius_increment
+ self.rotations = rotations
+
+ def generate(self, num_points=10, interval_seconds=10):
+ points = []
+ for i in range(num_points):
+ angle = 2 * math.pi * self.rotations * (i / num_points)
+ radius = self.initial_radius + self.radius_increment * i
+ delta_lat = radius * math.cos(angle)
+ delta_lon = radius * math.sin(angle)
+ lat = self.center_lat + delta_lat
+ lon = self.center_lon + delta_lon
+ alt = self._get_altitude()
+ time_delta = i * interval_seconds
+ points.append(self._create_track_point(lon, lat, alt, time_delta))
+ return points
+
+# --- Ownship-Relative Generator ---
+
+class OwnshipRelativeTrackGenerator(BaseTrackGenerator):
+ """
+ Generates a relative track based on an ownship trajectory.
+ Each generated point is offset from the corresponding ownship point by a specified separation.
+ """
+ def __init__(self, ownship_track, separation_distance, relative_direction="ahead", alt_offset=0, **kwargs):
+ super().__init__(center_lat=0, center_lon=0, center_alt=0, **kwargs)
+ self.ownship_track = ownship_track
+ self.separation_distance = separation_distance # in meters
+ self.relative_direction = relative_direction.lower()
+ self.alt_offset = alt_offset
+ self.start_time = datetime.now(timezone.utc)
+
+ @staticmethod
+ def compute_bearing(lat1, lon1, lat2, lon2):
+ lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2])
+ d_lon = lon2 - lon1
+ x = math.sin(d_lon) * math.cos(lat2)
+ y = math.cos(lat1) * math.sin(lat2) - math.sin(lat1) * math.cos(lat2) * math.cos(d_lon)
+ bearing = math.atan2(x, y)
+ if bearing < 0:
+ bearing += 2 * math.pi
+ return bearing
+
+ @staticmethod
+ def destination_point(lat, lon, bearing, distance):
+ R = 6371000.0 # Earth radius in meters.
+ lat_rad = math.radians(lat)
+ lon_rad = math.radians(lon)
+ angular_distance = distance / R
+ new_lat = math.asin(math.sin(lat_rad) * math.cos(angular_distance) +
+ math.cos(lat_rad) * math.sin(angular_distance) * math.cos(bearing))
+ new_lon = lon_rad + math.atan2(math.sin(bearing) * math.sin(angular_distance) * math.cos(lat_rad),
+ math.cos(angular_distance) - math.sin(lat_rad) * math.sin(new_lat))
+ return math.degrees(new_lat), math.degrees(new_lon)
+
+ def generate(self):
+ points = []
+ ownship = self.ownship_track
+ n = len(ownship)
+ previous_bearing = None
+ for i, pt in enumerate(ownship):
+ lat = pt.get("lat", pt.get("latitude"))
+ lon = pt.get("lon", pt.get("longitude"))
+ alt = pt.get("alt", pt.get("altitude"))
+ time_delta = pt.get("time", i * 10)
+ if i < n - 1:
+ next_pt = ownship[i + 1]
+ lat2 = next_pt.get("lat", next_pt.get("latitude"))
+ lon2 = next_pt.get("lon", next_pt.get("longitude"))
+ bearing = self.compute_bearing(lat, lon, lat2, lon2)
+ previous_bearing = bearing
+ elif previous_bearing is not None:
+ bearing = previous_bearing
+ else:
+ bearing = 0.0
+ if self.relative_direction == "ahead":
+ offset_bearing = bearing
+ offset_distance = self.separation_distance
+ elif self.relative_direction == "behind":
+ offset_bearing = bearing
+ offset_distance = -self.separation_distance
+ elif self.relative_direction == "opposite":
+ offset_bearing = (bearing + math.pi) % (2 * math.pi)
+ offset_distance = self.separation_distance
+ else:
+ offset_bearing = bearing
+ offset_distance = self.separation_distance
+ new_lat, new_lon = self.destination_point(lat, lon, offset_bearing, offset_distance)
+ new_alt = alt + self.alt_offset
+ track_point = self._create_track_point(new_lon, new_lat, new_alt, time_delta)
+ points.append(track_point)
+ return points
+
+
+
+class ArtificialTrackGenerator:
+ def __init__(self,
+ num_tracks=3,
+ track_generator_class='random', # Options: random, circular, elliptical, flyby, square, rectangle, zigzag, spiral
+ center_lat=33.75,
+ center_lon=-118.25,
+ center_alt=1000,
+ separation_distance=0.005, # For independent tracks (in degrees, approx. 0.005° ~ 555 m)
+ time_delay=30, # Time delay in seconds between tracks
+ min_alt=None,
+ max_alt=None,
+ fly_along=False, # If True, subsequent tracks are generated relative to the first (leader) track
+ ownship_trajectory_file=None, # Path to an ownship trajectory file (if fly_along is True)
+ ownship_format="json", # Format: "json" or "csv"
+ num_points=10,
+ interval_seconds=10):
+ self.num_tracks = num_tracks
+ self.track_generator_class = track_generator_class.lower()
+ self.center_lat = center_lat
+ self.center_lon = center_lon
+ self.center_alt = center_alt
+ self.separation_distance = separation_distance
+ self.time_delay = time_delay
+ self.min_alt = min_alt if min_alt is not None else center_alt
+ self.max_alt = max_alt if max_alt is not None else center_alt
+ self.fly_along = fly_along
+ self.initial_track_id = 10000 # 5-digit starting ID
+ self.ownship_trajectory_file = ownship_trajectory_file
+ self.ownship_format = ownship_format
+ self.num_points = num_points
+ self.interval_seconds = interval_seconds
+ self.tracks = [] # List of lists containing track point dictionaries
+
+ def _select_generator(self, **kwargs):
+ gen_class = self.track_generator_class
+ if gen_class == 'random':
+ return RandomTrackGenerator(**kwargs)
+ elif gen_class == 'circular':
+ return CircularTrackGenerator(**kwargs)
+ elif gen_class == 'elliptical':
+ return EllipticalTrackGenerator(**kwargs)
+ elif gen_class == 'flyby':
+ return FlybyTrackGenerator(**kwargs)
+ elif gen_class == 'square':
+ return SquareTrackGenerator(**kwargs)
+ elif gen_class == 'rectangle':
+ return RectangleTrackGenerator(**kwargs)
+ elif gen_class == 'zigzag':
+ return ZigzagTrackGenerator(**kwargs)
+ elif gen_class == 'spiral':
+ return SpiralTrackGenerator(**kwargs)
+ else:
+ return RandomTrackGenerator(**kwargs)
+
+ def generate_track(self, track_id, center_lat, center_lon, start_time):
+ kwargs = {
+ "center_lat": center_lat,
+ "center_lon": center_lon,
+ "center_alt": self.center_alt,
+ "min_alt": self.min_alt,
+ "max_alt": self.max_alt,
+ "track_id": track_id,
+ }
+ generator = self._select_generator(**kwargs)
+ generator.start_time = start_time
+ return generator.generate(num_points=self.num_points, interval_seconds=self.interval_seconds)
+
+ def generate(self):
+ base_start_time = datetime.now(timezone.utc)
+ tracks = []
+ if self.fly_along:
+ if self.ownship_trajectory_file:
+ ownship_track = load_ownship_trajectory(self.ownship_trajectory_file, self.ownship_format)
+ else:
+ ownship_track = self.generate_track(track_id=self.initial_track_id,
+ center_lat=self.center_lat,
+ center_lon=self.center_lon,
+ start_time=base_start_time)
+ tracks.append(ownship_track)
+ for i in range(1, self.num_tracks):
+ track_start_time = base_start_time + timedelta(seconds=i * self.time_delay)
+ kwargs = {
+ "ownship_track": ownship_track,
+ "separation_distance": self.separation_distance * 111000, # convert degrees to meters
+ "relative_direction": "ahead",
+ "alt_offset": 0,
+ "track_id": self.initial_track_id + i,
+ "source_id": "ARTIFICIAL",
+ "track_type": "Surveillance",
+ "altitude_reference": "MSL",
+ }
+ relative_generator = OwnshipRelativeTrackGenerator(**kwargs)
+ relative_generator.start_time = track_start_time
+ tracks.append(relative_generator.generate())
+ else:
+ for i in range(self.num_tracks):
+ offset_deg = self.separation_distance * i
+ track_center_lat = self.center_lat + offset_deg
+ track_center_lon = self.center_lon
+ track_start_time = base_start_time + timedelta(seconds=i * self.time_delay)
+ tracks.append(self.generate_track(track_id=self.initial_track_id + i,
+ center_lat=track_center_lat,
+ center_lon=track_center_lon,
+ start_time=track_start_time))
+ self.tracks = tracks
+ return tracks
+
+ def save_to_csv(self, filepath):
+ """
+ Save all generated tracks into a single CSV file.
+ """
+ if not self.tracks:
+ print("No tracks generated to save.")
+ return
+ fieldnames = [
+ "eventName", "trackId", "sourceId", "trackType", "timestamp", "data",
+ "altitudeReference", "sourceTrackId", "longitude", "latitude", "altitude",
+ "aglAltitude", "mslAltitude", "wgs84Altitude", "distance_m"
+ ]
+ with open(filepath, 'w', newline='') as csvfile:
+ writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
+ writer.writeheader()
+ for track in self.tracks:
+ for point in track:
+ writer.writerow(point)
+ print(f"Artificial tracks saved to {filepath}")
+
+# --- Example Usage ---
+if __name__ == "__main__":
+ # Example 1: Independent tracks using a "circular" maneuver.
+ generator1 = ArtificialTrackGenerator(
+ num_tracks=5,
+ track_generator_class='circular', # Options: random, circular, elliptical, flyby, square, rectangle, zigzag, spiral
+ center_lat=34.217411,
+ center_lon=-118.491081,
+ center_alt=1000,
+ separation_distance=0.005, # In degrees (approx. 555 m per 0.005° latitude)
+ time_delay=30, # 30-second delay between tracks
+ min_alt=900,
+ max_alt=1100,
+ fly_along=False, # Independent mode
+ num_points=20,
+ interval_seconds=10
+ )
+ generator1.generate()
+ generator1.save_to_csv("../outputs/output.csv")
+
+ # Example 2: Fly-along mode using an ownship trajectory from a JSON file.
+ # Uncomment and update the file path as needed.
+ # generator2 = ArtificialTrackGenerator(
+ # num_tracks=3,
+ # track_generator_class='zigzag', # Options: zigzag, spiral, etc.
+ # center_lat=33.75,
+ # center_lon=-118.25,
+ # center_alt=1000,
+ # separation_distance=0.005,
+ # time_delay=30,
+ # min_alt=900,
+ # max_alt=1100,
+ # fly_along=True,
+ # ownship_trajectory_file="ownship_trajectory.json",
+ # ownship_format="json",
+ # num_points=20,
+ # interval_seconds=10
+ # )
+ # generator2.generate()
+ # generator2.save_to_csv("artificial_tracks_flyalong.csv")
diff --git a/aerosim/src/aerosim/utils/config.py b/aerosim/src/aerosim/utils/config.py
new file mode 100644
index 0000000..b7ceb2d
--- /dev/null
+++ b/aerosim/src/aerosim/utils/config.py
@@ -0,0 +1,124 @@
+import json
+from pathlib import Path
+from typing import Dict, Any
+
+class Config:
+ """Configuration management class for Aerosim Utils."""
+
+ DEFAULT_CONFIG: Dict[str, Any] = {
+ # Directory paths
+ "DEFAULT_INPUT_DIR": "inputs",
+ "DEFAULT_OUTPUT_DIR": "outputs",
+ "DEFAULT_TRACKS_DIR": "tracks",
+ "DEFAULT_TRAJECTORIES_DIR": "trajectories",
+ "DEFAULT_VISUALIZATION_DIR": "visualization",
+ "DEFAULT_FOLIUM_DIR": "visualization/folium",
+ "DEFAULT_REPORTS_DIR": "reports",
+
+ # Default coordinates and parameters
+ "DEFAULT_CENTER_LAT": 34.217411,
+ "DEFAULT_CENTER_LON": -118.491081,
+ "DEFAULT_RADIUS_KM": 50.0,
+ "DEFAULT_ALTITUDE": 1000.0,
+ "DEFAULT_INTERVAL_SECONDS": 5.0,
+
+ # Scenario generation
+ "DEFAULT_SCENARIO_NAME": "auto_gen_scenario",
+ "DEFAULT_WORLD_NAME": "default_world",
+
+ # Visualization settings
+ "DEFAULT_VISUALIZATION_METHOD": "folium", # or 'kml'
+ "DEFAULT_MAP_CENTER": [34.217411, -118.491081],
+ "DEFAULT_MAP_ZOOM": 12,
+ "DEFAULT_TRACK_COLORS": ["red", "blue", "green", "purple", "orange"],
+
+ # Reporting settings
+ "DEFAULT_REPORT_FORMAT": "markdown", # or 'html'
+ "DEFAULT_PLOT_FORMAT": "png", # or 'svg', 'pdf'
+ "DEFAULT_PLOT_SIZE": [12, 8], # width, height in inches
+
+ # Sensor configuration
+ "DEFAULT_SENSOR_CONFIG": {
+ "type": "camera",
+ "fov": 90,
+ "resolution": [1920, 1080],
+ "update_rate": 30
+ }
+ }
+
+ _instance = None
+ _config: Dict[str, Any] = DEFAULT_CONFIG.copy()
+
+ def __new__(cls):
+ if cls._instance is None:
+ cls._instance = super(Config, cls).__new__(cls)
+ return cls._instance
+
+ @classmethod
+ def get(cls, key: str, default: Any = None) -> Any:
+ """Get a configuration value."""
+ return cls._config.get(key, default)
+
+ @classmethod
+ def set(cls, key: str, value: Any) -> None:
+ """Set a configuration value."""
+ cls._config[key] = value
+
+ @classmethod
+ def save_config(cls, config_file: str) -> None:
+ """Save current configuration to file."""
+ config_path = Path(config_file)
+ config_path.parent.mkdir(parents=True, exist_ok=True)
+ with open(config_path, 'w') as f:
+ json.dump(cls._config, f, indent=4)
+
+ @classmethod
+ def load_config(cls, config_file: str) -> None:
+ """Load configuration from file."""
+ config_path = Path(config_file)
+ if not config_path.exists():
+ raise FileNotFoundError(f"Configuration file not found: {config_file}")
+
+ with open(config_path, 'r') as f:
+ loaded_config = json.load(f)
+ cls._config.update(loaded_config)
+
+ @classmethod
+ def reset(cls) -> None:
+ """Reset configuration to defaults."""
+ cls._config = cls.DEFAULT_CONFIG.copy()
+
+ @classmethod
+ def validate_config(cls) -> bool:
+ """Validate the current configuration."""
+ try:
+ # Validate paths
+ for key in cls._config:
+ if key.endswith("_DIR"):
+ path = Path(cls._config[key])
+ if not path.is_absolute():
+ # Convert to absolute path relative to package root
+ cls._config[key] = str(Path(__file__).parent.parent / path)
+
+ # Validate numeric values
+ assert isinstance(cls._config["DEFAULT_CENTER_LAT"], (int, float))
+ assert isinstance(cls._config["DEFAULT_CENTER_LON"], (int, float))
+ assert isinstance(cls._config["DEFAULT_RADIUS_KM"], (int, float))
+ assert isinstance(cls._config["DEFAULT_ALTITUDE"], (int, float))
+ assert isinstance(cls._config["DEFAULT_INTERVAL_SECONDS"], (int, float))
+
+ # Validate visualization settings
+ assert cls._config["DEFAULT_VISUALIZATION_METHOD"] in ["folium", "kml"]
+ assert len(cls._config["DEFAULT_MAP_CENTER"]) == 2
+ assert isinstance(cls._config["DEFAULT_MAP_ZOOM"], int)
+ assert isinstance(cls._config["DEFAULT_TRACK_COLORS"], list)
+
+ # Validate reporting settings
+ assert cls._config["DEFAULT_REPORT_FORMAT"] in ["markdown", "html"]
+ assert cls._config["DEFAULT_PLOT_FORMAT"] in ["png", "svg", "pdf"]
+ assert len(cls._config["DEFAULT_PLOT_SIZE"]) == 2
+
+ return True
+ except (AssertionError, KeyError) as e:
+ print(f"Configuration validation failed: {str(e)}")
+ return False
diff --git a/aerosim/src/aerosim/utils/conversion.py b/aerosim/src/aerosim/utils/conversion.py
new file mode 100644
index 0000000..c4150b8
--- /dev/null
+++ b/aerosim/src/aerosim/utils/conversion.py
@@ -0,0 +1,216 @@
+import math
+import csv
+import json
+import random
+from datetime import datetime
+from collections import defaultdict
+
+def parse_timestamp(ts_str):
+ """
+ Parse an ISO8601 timestamp that may or may not include timezone information.
+
+ Examples:
+ "2024-07-25T18:21:11.337Z" or "2024-07-25T18:23:18.35Z" or "2024-07-25T18:27:22Z"
+
+ If timezone info is missing, the time is assumed to be in UTC.
+ """
+ from datetime import timezone
+ is_utc = False
+ # Remove trailing 'Z' and note that time is in UTC.
+ if ts_str.endswith("Z"):
+ ts_str = ts_str[:-1]
+ is_utc = True
+ # Attempt to remove colon from timezone offset if present.
+ if ("+" in ts_str or "-" in ts_str[19:]):
+ pos_plus = ts_str.rfind("+")
+ pos_minus = ts_str.rfind("-")
+ pos = max(pos_plus, pos_minus)
+ if pos != -1:
+ tz_part = ts_str[pos:]
+ tz_part = tz_part.replace(":", "")
+ ts_str = ts_str[:pos] + tz_part
+ # First, try parsing with timezone (%z)
+ try:
+ return datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S.%f%z")
+ except ValueError:
+ try:
+ # Try parsing with timezone but without milliseconds
+ return datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S%z")
+ except ValueError:
+ try:
+ # Try parsing without timezone but with milliseconds
+ dt = datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S.%f")
+ if is_utc:
+ dt = dt.replace(tzinfo=timezone.utc)
+ return dt
+ except ValueError:
+ # Fallback: try parsing without timezone and without milliseconds
+ dt = datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S")
+ if is_utc:
+ dt = dt.replace(tzinfo=timezone.utc)
+ return dt
+
+
+def convert_json_to_csv(input_json_path, output_csv_path):
+ """
+ Convert raw JSON data into a flattened CSV file.
+
+ Args:
+ input_json_path (str): Path to the input JSON file.
+ output_csv_path (str): Path to save the CSV file.
+ """
+ with open(input_json_path, 'r') as json_file:
+ data = json.load(json_file)
+
+ rows = []
+ for item in data:
+ # Get event-level info
+ event_name = item.get("eventName")
+ track = item.get("surveillanceTrack", {})
+
+ # Extract track details
+ row = {
+ "eventName": event_name,
+ "trackId": track.get("trackId"),
+ "sourceId": track.get("sourceId"),
+ "trackType": track.get("trackType"),
+ "timestamp": track.get("timestamp"),
+ "data": track.get("data"),
+ "altitudeReference": track.get("altitudeReference"),
+ "sourceTrackId": track.get("sourceTrackId")
+ }
+ # Extract nested reference location
+ ref_loc = track.get("referenceLocation", {})
+ row.update({
+ "longitude": ref_loc.get("longitude"),
+ "latitude": ref_loc.get("latitude"),
+ "altitude": ref_loc.get("altitude"),
+ "aglAltitude": ref_loc.get("aglAltitude"),
+ "mslAltitude": ref_loc.get("mslAltitude"),
+ "wgs84Altitude": ref_loc.get("wgs84Altitude")
+ })
+ rows.append(row)
+
+ fieldnames = [
+ "eventName", "trackId", "sourceId", "trackType", "timestamp", "data",
+ "altitudeReference", "sourceTrackId", "longitude", "latitude",
+ "altitude", "aglAltitude", "mslAltitude", "wgs84Altitude"
+ ]
+
+ with open(output_csv_path, 'w', newline='') as csv_file:
+ writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
+ writer.writeheader()
+ writer.writerows(rows)
+
+ print(f"CSV file has been created as '{output_csv_path}'.")
+
+def filter_tracks(input_csv_path: str,
+ filtered_csv_path: str,
+ tracks_folder: str,
+ ref_lat: float,
+ ref_lon: float,
+ ref_range_m: float = 10000) -> None:
+ """
+ Filter track data based on distance from a reference point and group by track.
+
+ Args:
+ input_csv_path (str): Path to the input CSV file.
+ filtered_csv_path (str): Path for the overall filtered CSV file.
+ tracks_folder (str): Directory where individual track CSV files will be saved.
+ ref_lat (float): Reference latitude.
+ ref_lon (float): Reference longitude.
+ ref_range_m (float, optional): Maximum allowed distance in meters. Defaults to 10000.
+ """
+ import pandas as pd
+ import geopy.distance
+
+ df = pd.read_csv(input_csv_path)
+ df = df.dropna(subset=['latitude', 'longitude'])
+
+ ref_coords = (ref_lat, ref_lon)
+
+ def calculate_distance(row: pd.Series) -> float:
+ point_coords = (row['latitude'], row['longitude'])
+ return geopy.distance.geodesic(ref_coords, point_coords).meters
+
+ df['distance_m'] = df.apply(calculate_distance, axis=1)
+ filtered_df = df[df['distance_m'] < ref_range_m]
+ filtered_df.to_csv(filtered_csv_path, index=False)
+ print(f"Filtered data saved to '{filtered_csv_path}'")
+
+ import os
+ os.makedirs(tracks_folder, exist_ok=True)
+ for track_id, group in filtered_df.groupby("trackId"):
+ # Format track ID as a 5-digit number
+ try:
+ tid = int(float(track_id))
+ except ValueError:
+ tid = track_id
+ track_path = os.path.join(tracks_folder, f"track_{tid:05d}.csv")
+ group.to_csv(track_path, index=False)
+ print(f"Track {track_id} data saved to '{track_path}'")
+
+def process_csv(csv_path):
+ """
+ Process a CSV file of ADS-B track data:
+ - Group rows by trackId.
+ - Compute relative timestamps (starting at zero for each track).
+ - Return a dict mapping trackId to trajectory list.
+ """
+ tracks = defaultdict(list)
+ with open(csv_path, newline="") as csvfile:
+ reader = csv.DictReader(csvfile)
+ for row in reader:
+ track_id = row["trackId"]
+ try:
+ ts = parse_timestamp(row["timestamp"])
+ except ValueError as e:
+ print(f"Error parsing timestamp {row['timestamp']}: {e}")
+ continue
+ row["parsed_ts"] = ts
+ tracks[track_id].append(row)
+ trajectories = {}
+ for track_id, rows in tracks.items():
+ rows.sort(key=lambda x: x["parsed_ts"])
+ base_ts = rows[0]["parsed_ts"]
+ trajectory = []
+ for row in rows:
+ delta = (row["parsed_ts"] - base_ts).total_seconds()
+ lat = float(row["latitude"])
+ lon = float(row["longitude"])
+ try:
+ height = float(row["mslAltitude"])
+ except (ValueError, TypeError):
+ height = float(row["altitude"])
+ trajectory.append({
+ "time": delta,
+ "lat": lat,
+ "lon": lon,
+ "alt": height
+ })
+ trajectories[track_id] = trajectory
+ return trajectories
+
+def convert_tracks_to_json(input_folder, output_folder):
+ """
+ Convert all CSV track files in input_folder to Aerosim trajectory JSON files.
+
+ Args:
+ input_folder (str): Directory containing CSV track files.
+ output_folder (str): Directory where trajectory JSON files will be saved.
+ """
+ import os
+ os.makedirs(output_folder, exist_ok=True)
+ for filename in os.listdir(input_folder):
+ if filename.lower().endswith(".csv"):
+ csv_path = os.path.join(input_folder, filename)
+ trajectories = process_csv(csv_path)
+ for track_id, trajectory in trajectories.items():
+ try:
+ track_num = int(float(track_id))
+ except ValueError:
+ track_num = track_id
+ output_filename = os.path.join(output_folder, f"{track_num:05d}_trajectory.json")
+ with open(output_filename, "w") as jsonfile:
+ json.dump(trajectory, jsonfile, indent=2)
+ print(f"Saved trajectory for track {track_id} to {output_filename}")
diff --git a/aerosim/src/aerosim/utils/process_workflows.py b/aerosim/src/aerosim/utils/process_workflows.py
new file mode 100644
index 0000000..c47be24
--- /dev/null
+++ b/aerosim/src/aerosim/utils/process_workflows.py
@@ -0,0 +1,314 @@
+#!/usr/bin/env python3
+"""
+Process Workflows for Utils
+
+This module defines functions to run complete processing workflows for three use cases:
+ 1. OpenADSB workflow (real ADS-B data processing)
+ 2. Artificial track generation workflow
+ 3. Google Earth KML workflow
+
+Each workflow performs the following steps:
+ - Convert raw data to CSV (or use an existing CSV file)
+ - Filter or process the CSV into trajectory JSON files
+ - Create a simulation scenario JSON from the trajectories
+ - Visualize the tracks (e.g. as a combined Folium map)
+ - Generate scenario reports and trajectory plots
+"""
+
+import json
+from pathlib import Path
+from datetime import timedelta, timezone, datetime
+
+from utils import (
+ conversion,
+ scenario_generator,
+ visualization,
+ ArtificialTrackGenerator,
+ scenario_report,
+)
+from utils.tracks_from_map import parse_kml, generate_csv_from_tracks
+
+# Set up default directories (you may also load these from a configuration file)
+DEFAULT_INPUT_DIR = Path("../inputs")
+DEFAULT_OUTPUT_DIR = Path("../outputs")
+DEFAULT_TRACKS_DIR = Path("../tracks")
+DEFAULT_TRAJECTORIES_DIR = Path("../trajectories")
+DEFAULT_VISUALIZATION_DIR = Path("../visualization")
+DEFAULT_FOLIUM_DIR = Path("../visualization/folium")
+DEFAULT_REPORTS_DIR = Path("../reports")
+
+# Create folders if they do not exist
+for folder in [DEFAULT_OUTPUT_DIR, DEFAULT_TRACKS_DIR, DEFAULT_TRAJECTORIES_DIR,
+ DEFAULT_VISUALIZATION_DIR, DEFAULT_FOLIUM_DIR, DEFAULT_REPORTS_DIR]:
+ folder.mkdir(parents=True, exist_ok=True)
+
+
+def process_openadsb_workflow(
+ input_json: Path = None,
+ center_lat: float = 34.217411,
+ center_lon: float = -118.491081,
+ radius_km: float = 5,
+ num_vehicles: int = None,
+ interval: int = 10,
+ generate_report: bool = True,
+ report_output: Path = None,
+ plot_trajectories: bool = True,
+ plot_output: Path = None,
+) -> dict:
+ """
+ Process workflow for ADS-B data.
+
+ Steps:
+ 1. Convert raw ADS-B JSON data to CSV.
+ 2. Filter tracks around a reference point.
+ 3. Convert filtered track CSV files into trajectory JSON files.
+ 4. Generate a simulation scenario JSON from trajectories.
+ 5. Visualize the tracks.
+ 6. Generate scenario report and trajectory plot (optional).
+
+ Args:
+ input_json: Path to the input ADS-B JSON file. If not provided, the first JSON in inputs/ is used.
+ center_lat: Reference latitude for filtering.
+ center_lon: Reference longitude for filtering.
+ radius_km: Search radius in kilometers.
+ num_vehicles: (Optional) Number of vehicles to include in the scenario.
+ interval: Time interval (seconds) between points for the scenario.
+ generate_report: Whether to generate a markdown report.
+ report_output: Path for the markdown report. If None, uses default location.
+ plot_trajectories: Whether to generate a trajectory plot.
+ plot_output: Path for the trajectory plot. If None, uses default location.
+
+ Returns:
+ The generated scenario as a Python dictionary.
+ """
+ # Use first JSON file in DEFAULT_INPUT_DIR if input not provided.
+ if input_json is None:
+ json_files = list(DEFAULT_INPUT_DIR.glob("*.json"))
+ if not json_files:
+ raise FileNotFoundError("No JSON files found in the inputs folder.")
+ input_json = json_files[0]
+ print(f"No input JSON provided. Using {input_json}")
+
+ csv_path = DEFAULT_OUTPUT_DIR / "output.csv"
+ conversion.convert_json_to_csv(str(input_json), str(csv_path))
+
+ filtered_csv = DEFAULT_OUTPUT_DIR / "filtered_tracks.csv"
+ conversion.filter_tracks(
+ input_csv_path=str(csv_path),
+ filtered_csv_path=str(filtered_csv),
+ tracks_folder=str(DEFAULT_TRACKS_DIR),
+ ref_lat=center_lat,
+ ref_lon=center_lon,
+ ref_range_m=radius_km * 1000,
+ )
+
+ # Convert filtered track CSV files to trajectory JSON files
+ conversion.convert_tracks_to_json(str(DEFAULT_TRACKS_DIR), str(DEFAULT_TRAJECTORIES_DIR))
+
+ # Generate scenario JSON from trajectory JSON files
+ scenario_json_file = DEFAULT_OUTPUT_DIR / "scenario.json"
+ scenario_generator.generate_scenario_json(str(DEFAULT_TRAJECTORIES_DIR), str(scenario_json_file))
+
+ # Visualize the tracks (Folium map)
+ visualization.visualize_folder(str(DEFAULT_TRACKS_DIR), str(DEFAULT_VISUALIZATION_DIR / "folium" / "combined_map.html"), method="folium")
+
+ with open(scenario_json_file, 'r') as f:
+ scenario_data = json.load(f)
+
+ # Generate report and plot if requested
+ if generate_report:
+ if report_output is None:
+ report_output = DEFAULT_REPORTS_DIR / "scenario_report.md"
+ report_md = scenario_report.generate_markdown_report(scenario_data)
+ scenario_report.save_markdown_report(report_md, str(report_output))
+ print(f"Scenario report saved to {report_output}")
+
+ if plot_trajectories:
+ if plot_output is None:
+ plot_output = DEFAULT_REPORTS_DIR / "trajectories.png"
+ scenario_report.plot_trajectories(scenario_data, str(plot_output))
+ print(f"Trajectory plot saved to {plot_output}")
+
+ return scenario_data
+
+
+def process_artificial_workflow(
+ maneuver: str = "elliptical",
+ num_tracks: int = 3,
+ center_lat: float = 34.217411,
+ center_lon: float = -118.491081,
+ center_alt: float = 1000,
+ separation: float = 0.005,
+ time_delay: int = 30,
+ num_points: int = 20,
+ interval_seconds: int = 10,
+ generate_report: bool = True,
+ report_output: Path = None,
+ plot_trajectories: bool = True,
+ plot_output: Path = None,
+) -> dict:
+ """
+ Process workflow for generating artificial tracks.
+
+ Steps:
+ 1. Generate artificial tracks using the specified maneuver.
+ 2. Save the tracks to a CSV file.
+ 3. Convert the CSV (with multiple track IDs) to trajectory JSON files.
+ 4. Generate a simulation scenario JSON from trajectories.
+ 5. Visualize the tracks.
+ 6. Generate scenario report and trajectory plot (optional).
+
+ Args:
+ maneuver: Track pattern type (random, circular, elliptical, flyby, square, rectangle, zigzag, spiral).
+ num_tracks: Number of tracks to generate.
+ center_lat, center_lon, center_alt: Center coordinates and altitude.
+ separation: Separation distance in degrees between independent tracks.
+ time_delay: Time delay in seconds between tracks.
+ num_points: Number of points per track.
+ interval_seconds: Time interval in seconds between points.
+ generate_report: Whether to generate a markdown report.
+ report_output: Path for the markdown report. If None, uses default location.
+ plot_trajectories: Whether to generate a trajectory plot.
+ plot_output: Path for the trajectory plot. If None, uses default location.
+
+ Returns:
+ The generated scenario as a Python dictionary.
+ """
+ # Instantiate the artificial track generator with given parameters.
+ generator = ArtificialTrackGenerator(
+ num_tracks=num_tracks,
+ track_generator_class=maneuver,
+ center_lat=center_lat,
+ center_lon=center_lon,
+ center_alt=center_alt,
+ separation_distance=separation,
+ time_delay=time_delay,
+ min_alt=center_alt * 0.9,
+ max_alt=center_alt * 1.1,
+ fly_along=False,
+ num_points=num_points,
+ interval_seconds=interval_seconds,
+ )
+ generator.generate()
+ artificial_csv = DEFAULT_OUTPUT_DIR / "artificial_tracks.csv"
+ generator.save_to_csv(str(artificial_csv))
+
+ # Process the artificial CSV file to generate trajectory JSON files.
+ conversion.convert_tracks_to_json(str(DEFAULT_OUTPUT_DIR), str(DEFAULT_TRAJECTORIES_DIR))
+
+ scenario_json_file = DEFAULT_OUTPUT_DIR / "scenario_artificial.json"
+ scenario_generator.generate_scenario_json(str(DEFAULT_TRAJECTORIES_DIR), str(scenario_json_file))
+
+ # Visualize the artificial tracks
+ visualization.visualize_folder(str(DEFAULT_OUTPUT_DIR), str(DEFAULT_VISUALIZATION_DIR / "folium" / "combined_map_artificial.html"), method="folium")
+
+ with open(scenario_json_file, 'r') as f:
+ scenario_data = json.load(f)
+
+ # Generate report and plot if requested
+ if generate_report:
+ if report_output is None:
+ report_output = DEFAULT_REPORTS_DIR / "scenario_report_artificial.md"
+ report_md = scenario_report.generate_markdown_report(scenario_data)
+ scenario_report.save_markdown_report(report_md, str(report_output))
+ print(f"Scenario report saved to {report_output}")
+
+ if plot_trajectories:
+ if plot_output is None:
+ plot_output = DEFAULT_REPORTS_DIR / "trajectories_artificial.png"
+ scenario_report.plot_trajectories(scenario_data, str(plot_output))
+ print(f"Trajectory plot saved to {plot_output}")
+
+ return scenario_data
+
+
+def process_kml_workflow(
+ input_kml: Path,
+ interval: int = 10,
+ generate_report: bool = True,
+ report_output: Path = None,
+ plot_trajectories: bool = True,
+ plot_output: Path = None,
+) -> dict:
+ """
+ Process workflow for tracks created in Google Earth (KML).
+
+ Steps:
+ 1. Convert the KML file to CSV using tracks-from-map functionality.
+ 2. Convert the resulting CSV to trajectory JSON files.
+ 3. Generate a simulation scenario JSON from the trajectories.
+ 4. Visualize the tracks.
+ 5. Generate scenario report and trajectory plot (optional).
+
+ Args:
+ input_kml: Path to the KML file exported from Google Earth.
+ interval: Time interval in seconds between track points.
+ generate_report: Whether to generate a markdown report.
+ report_output: Path for the markdown report. If None, uses default location.
+ plot_trajectories: Whether to generate a trajectory plot.
+ plot_output: Path for the trajectory plot. If None, uses default location.
+
+ Returns:
+ The generated scenario as a Python dictionary.
+ """
+ from utils.tracks_from_map import parse_kml, generate_csv_from_tracks
+
+ # Parse the KML file to get track data
+ tracks = parse_kml(str(input_kml))
+ if not tracks:
+ raise ValueError("No tracks found in the provided KML file.")
+
+ kml_csv = DEFAULT_OUTPUT_DIR / "kml_output.csv"
+ generate_csv_from_tracks(tracks, str(kml_csv), interval_seconds=interval)
+
+ # Convert the generated CSV to trajectory JSON files.
+ conversion.convert_tracks_to_json(str(DEFAULT_OUTPUT_DIR), str(DEFAULT_TRAJECTORIES_DIR))
+
+ scenario_json_file = DEFAULT_OUTPUT_DIR / "scenario_kml.json"
+ scenario_generator.generate_scenario_json(str(DEFAULT_TRAJECTORIES_DIR), str(scenario_json_file))
+
+ # Visualize the tracks
+ visualization.visualize_folder(str(DEFAULT_OUTPUT_DIR), str(DEFAULT_VISUALIZATION_DIR / "folium" / "combined_map_kml.html"), method="folium")
+
+ with open(scenario_json_file, 'r') as f:
+ scenario_data = json.load(f)
+
+ # Generate report and plot if requested
+ if generate_report:
+ if report_output is None:
+ report_output = DEFAULT_REPORTS_DIR / "scenario_report_kml.md"
+ report_md = scenario_report.generate_markdown_report(scenario_data)
+ scenario_report.save_markdown_report(report_md, str(report_output))
+ print(f"Scenario report saved to {report_output}")
+
+ if plot_trajectories:
+ if plot_output is None:
+ plot_output = DEFAULT_REPORTS_DIR / "trajectories_kml.png"
+ scenario_report.plot_trajectories(scenario_data, str(plot_output))
+ print(f"Trajectory plot saved to {plot_output}")
+
+ return scenario_data
+
+
+def main():
+ """
+ Run all three workflows sequentially:
+ 1. OpenADSB Workflow
+ 2. Artificial Track Workflow
+ 3. Google Earth KML Workflow
+ """
+ # print("=== Running OpenADSB Workflow ===")
+ # openadsb_scenario = process_openadsb_workflow()
+ # print("OpenADSB Scenario generated.")
+
+ print("\n=== Running Artificial Track Workflow ===")
+ artificial_scenario = process_artificial_workflow()
+ print("Artificial Track Scenario generated.")
+
+
+ # Optionally, save or further process the scenarios.
+ print("\nAll workflows completed.")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/aerosim/src/aerosim/utils/requirements.txt b/aerosim/src/aerosim/utils/requirements.txt
new file mode 100644
index 0000000..1216526
--- /dev/null
+++ b/aerosim/src/aerosim/utils/requirements.txt
@@ -0,0 +1,14 @@
+pandas>=2.2.3
+geopy>=2.4.1
+folium>=0.19.5
+matplotlib>=3.10.1
+numpy>=2.2.3
+click>=8.1.0
+simplekml>=1.3.1
+requests>=2.31.0
+python-dateutil>=2.8.2
+tqdm>=4.66.1
+pyyaml>=6.0.1
+geopandas>=0.14.1
+contextily>=1.4.0
+shapely>=2.0.2
\ No newline at end of file
diff --git a/aerosim/src/aerosim/utils/scenario_generator.py b/aerosim/src/aerosim/utils/scenario_generator.py
new file mode 100644
index 0000000..a4df922
--- /dev/null
+++ b/aerosim/src/aerosim/utils/scenario_generator.py
@@ -0,0 +1,251 @@
+import os
+import json
+
+def generate_scenario_json(trajectories_folder: str = "trajectories",
+ output_file: str = "auto_gen_scenario.json",
+ fmu_model_path: str = "dev_scripts/fmu/trajectory_follower_fmu_model.fmu",
+ sensor_configs: list = None,
+ add_sensors_to_vehicles: list = None,
+ weather_preset: str = "Cloudy",
+ renderer_viewport_config: dict = {"active_camera": "rgb_camera_0"}) -> None:
+ """
+ Generate a simulation scenario JSON using the new scenario schema.
+
+ The generated scenario includes:
+ - A clock with fixed step size.
+ - An orchestrator with sync topics.
+ - A world with an origin based on the first waypoint, updated weather, actors with trajectory visualization,
+ and sensors attached as specified.
+ - A primary renderer using the provided sensor configuration.
+ - FMU models for trajectory following per actor and additional sensor FMU models for the first actor.
+
+ Args:
+ trajectories_folder (str): Directory containing trajectory JSON files.
+ output_file (str): Output file path for the generated scenario JSON.
+ fmu_model_path (str): Path to the trajectory follower FMU model file.
+ sensor_configs (list): List of sensor configurations to attach (default provided if None).
+ add_sensors_to_vehicles (list): List of vehicle names to which sensors should be attached.
+ If None or empty, sensors are attached to the first actor encountered.
+ weather_preset (str): Weather preset for the simulation world.
+ renderer_viewport_config (dict): Configuration for the primary renderer's viewport.
+ """
+ if sensor_configs is None:
+ sensor_configs = [
+ {
+ "sensor_name": "rgb_camera_0",
+ "type": "sensors/cameras/rgb_camera", # updated to match sample scenario
+ "transform": {
+ "translation": [-15.0, 0.0, -2.0],
+ "rotation": [0.0, -10.0, 0.0]
+ },
+ "parameters": {
+ "resolution": [1920, 1080],
+ "tick_rate": 0.02,
+ "frame_rate": 30,
+ "fov": 90,
+ "near_clip": 0.1,
+ "far_clip": 1000.0,
+ "capture_enabled": False
+ }
+ }
+ ]
+ if add_sensors_to_vehicles is None:
+ add_sensors_to_vehicles = []
+
+ # Base scenario structure
+ scenario = {
+ "description": "Multi-intruder scenario generated from trajectories.",
+ "clock": {
+ "step_size_ms": 20,
+ "pace_1x_scale": True
+ },
+ "orchestrator": {
+ "sync_topics": [] # updated schema: no create_topics
+ },
+ "world": {
+ "update_interval_ms": 20,
+ "origin": {},
+ "weather": {
+ "preset": weather_preset
+ },
+ "actors": [],
+ "sensors": []
+ },
+ "renderers": [{
+ "renderer_id": "0",
+ "role": "primary",
+ "sensors": [sensor['sensor_name'] for sensor in sensor_configs],
+ "viewport_config": renderer_viewport_config
+ }],
+ "fmu_models": []
+ }
+
+ # Retrieve and sort trajectory files for consistent ordering
+ trajectory_files = [f for f in os.listdir(trajectories_folder) if f.endswith('_trajectory.json')]
+ trajectory_files.sort()
+
+ for idx, traj_file in enumerate(trajectory_files, start=1):
+ track_id = traj_file.split('_')[0]
+ traj_path = os.path.join(trajectories_folder, traj_file)
+ with open(traj_path, 'r') as file:
+ waypoints = json.load(file)
+ if not waypoints:
+ continue
+ first_waypoint = waypoints[0]
+ if idx == 1:
+ # Use the first waypoint to define the world origin (altitude defaults to reference value if not provided)
+ scenario['world']['origin'] = {
+ "latitude": first_waypoint['lat'],
+ "longitude": first_waypoint['lon'],
+ "altitude": first_waypoint.get('alt', 116.09)
+ }
+ actor_name = f"actor{track_id}"
+ vehicle_topic = f"aerosim.{actor_name}.vehicle_state"
+ trajectory_vis_topic = f"aerosim.{actor_name}.trajectory_visualization"
+ effector_topic = f"aerosim.{actor_name}.effector1.state"
+
+ # Add sync topic for the actor
+ scenario['orchestrator']['sync_topics'].append({
+ "topic": vehicle_topic,
+ "interval_ms": 20
+ })
+
+ # Create actor configuration with updated asset path, effector state, and added trajectory visualization
+ actor = {
+ "actor_name": actor_name,
+ "actor_asset": "vehicles/generic_airplane/generic_airplane",
+ "parent": "",
+ "description": f"Trajectory follower {track_id}",
+ "transform": {
+ "position": [0.0, 0.0, 0.0],
+ "rotation": [0.0, 0.0, 0.0],
+ "scale": [1.0, 1.0, 1.0]
+ },
+ "state": {
+ "msg_type": "aerosim::types::VehicleState",
+ "topic": vehicle_topic
+ },
+ "effectors": [{
+ "id": "propeller_front",
+ "relative_path": "generic_airplane/propeller",
+ "transform": {
+ "translation": [3.1, 0.0, 0.0],
+ "rotation": [0.0, -90.0, 0.0],
+ "scale": [1.0, 1.0, 1.0]
+ },
+ "state": {
+ "msg_type": "aerosim::types::EffectorState",
+ "topic": effector_topic
+ }
+ }],
+ "trajectory_visualization": {
+ "msg_type": "aerosim::types::TrajectoryVisualization",
+ "topic": trajectory_vis_topic
+ }
+ }
+ scenario['world']['actors'].append(actor)
+
+ # Attach sensors to vehicles if specified (default to first actor)
+ if idx == 1 and not add_sensors_to_vehicles:
+ add_sensors_to_vehicles.append(actor_name)
+
+ if actor_name in add_sensors_to_vehicles:
+ for sensor in sensor_configs:
+ sensor_config = sensor.copy()
+ sensor_config['parent'] = actor_name
+ scenario['world']['sensors'].append(sensor_config)
+
+ # Add trajectory follower FMU model for the actor using the updated initial values
+ traj_follower_fmu = {
+ "id": f"trajectory_follower_{track_id}",
+ "fmu_model_path": fmu_model_path,
+ "component_input_topics": [],
+ "component_output_topics": [
+ {
+ "msg_type": "aerosim::types::VehicleState",
+ "topic": vehicle_topic
+ },
+ {
+ "msg_type": "aerosim::types::TrajectoryVisualization",
+ "topic": trajectory_vis_topic
+ }
+ ],
+ "fmu_aux_input_mapping": {},
+ "fmu_aux_output_mapping": {},
+ "fmu_initial_vals": {
+ "waypoints_json_path": os.path.join(trajectories_folder, traj_file),
+ "display_future_trajectory": True,
+ "display_past_trajectory": True,
+ "highlight_user_defined_waypoints": True,
+ "number_of_future_waypoints": 1,
+ "use_linear_interpolation": False,
+ "time_step_in_seconds": 0.01,
+ "curvature_roll_factor": 1.0,
+ "max_roll_rate_deg_per_second": 10.0
+ }
+ }
+ scenario['fmu_models'].append(traj_follower_fmu)
+
+ # For the first actor, also add sensor FMU models (GNSS, ADSB, IMU)
+ if idx == 1:
+ # GNSS publisher
+ gnss_fmu = {
+ "id": "gnss_publisher",
+ "fmu_model_path": "fmu/gnss_sensor_fmu_model.fmu",
+ "component_type": "sensor",
+ "component_input_topics": [
+ {
+ "msg_type": "aerosim::types::VehicleState",
+ "topic": vehicle_topic
+ }
+ ],
+ "component_output_topics": [],
+ "fmu_aux_input_mapping": {},
+ "fmu_aux_output_mapping": {},
+ "fmu_initial_vals": {
+ "output_gnss_topic_name": f"aerosim.{actor_name}.sensor.gnss"
+ }
+ }
+ scenario['fmu_models'].append(gnss_fmu)
+ # ADSB publisher
+ adsb_fmu = {
+ "id": "adsb_publisher",
+ "fmu_model_path": "fmu/adsb_sensor_fmu_model.fmu",
+ "component_type": "sensor",
+ "component_input_topics": [
+ {
+ "msg_type": "aerosim::types::VehicleState",
+ "topic": vehicle_topic
+ }
+ ],
+ "component_output_topics": [],
+ "fmu_aux_input_mapping": {},
+ "fmu_aux_output_mapping": {},
+ "fmu_initial_vals": {
+ "output_adsb_topic_name": f"aerosim.{actor_name}.sensor.adsb"
+ }
+ }
+ scenario['fmu_models'].append(adsb_fmu)
+ # IMU publisher
+ imu_fmu = {
+ "id": "imu_publisher",
+ "fmu_model_path": "fmu/imu_sensor_fmu_model.fmu",
+ "component_type": "sensor",
+ "component_input_topics": [
+ {
+ "msg_type": "aerosim::types::VehicleState",
+ "topic": vehicle_topic
+ }
+ ],
+ "component_output_topics": [],
+ "fmu_aux_input_mapping": {},
+ "fmu_aux_output_mapping": {},
+ "fmu_initial_vals": {
+ "output_imu_topic_name": f"aerosim.{actor_name}.sensor.imu"
+ }
+ }
+ scenario['fmu_models'].append(imu_fmu)
+
+ with open(output_file, 'w') as outfile:
+ json.dump(scenario, outfile, indent=4)
+ print(f"Scenario JSON created at {output_file}")
diff --git a/aerosim/src/aerosim/utils/scenario_report.py b/aerosim/src/aerosim/utils/scenario_report.py
new file mode 100644
index 0000000..041c137
--- /dev/null
+++ b/aerosim/src/aerosim/utils/scenario_report.py
@@ -0,0 +1,248 @@
+import json
+import os
+import sys
+
+def generate_markdown_report(scenario):
+ lines = []
+ lines.append("# Simulation Scenario Report")
+ lines.append("")
+ # Overview
+ lines.append("## Overview")
+ lines.append("")
+ lines.append(f"**Description:** {scenario.get('description', 'No description provided.')}")
+ lines.append("")
+
+ # Clock
+ clock = scenario.get("clock", {})
+ lines.append("## Clock Settings")
+ lines.append("")
+ lines.append(f"- **Step Size (ms):** {clock.get('step_size_ms', 'N/A')}")
+ lines.append(f"- **Pace 1x Scale:** {clock.get('pace_1x_scale', 'N/A')}")
+ lines.append("")
+
+ # Orchestrator
+ orch = scenario.get("orchestrator", {})
+ lines.append("## Orchestrator")
+ lines.append("")
+ sync_topics = orch.get("sync_topics", [])
+ lines.append("- **Sync Topics:**")
+ for topic in sync_topics:
+ lines.append(f" - **Topic:** `{topic.get('topic', 'N/A')}`, **Interval (ms):** {topic.get('interval_ms', 'N/A')}")
+ lines.append(f"- **Output Data File:** `{orch.get('output_sim_data_file', 'N/A')}`")
+ lines.append("")
+
+ # World Settings
+ world = scenario.get("world", {})
+ lines.append("## World Settings")
+ lines.append("")
+ lines.append(f"- **Update Interval (ms):** {world.get('update_interval_ms', 'N/A')}")
+ origin = world.get("origin", {})
+ lines.append("- **Origin:**")
+ lines.append(f" - **Latitude:** {origin.get('latitude', 'N/A')}")
+ lines.append(f" - **Longitude:** {origin.get('longitude', 'N/A')}")
+ lines.append(f" - **Altitude:** {origin.get('altitude', 'N/A')}")
+ weather = world.get("weather", {})
+ lines.append(f"- **Weather Preset:** {weather.get('preset', 'N/A')}")
+ lines.append("")
+
+ # Actors
+ actors = world.get("actors", [])
+ lines.append("## Actors")
+ lines.append("")
+ if actors:
+ for actor in actors:
+ lines.append(f"### Actor: {actor.get('actor_name', 'Unnamed')}")
+ lines.append(f"- **Asset:** {actor.get('actor_asset', 'N/A')}")
+ lines.append(f"- **Description:** {actor.get('description', 'No description')}")
+ transform = actor.get("transform", {})
+ lines.append(f"- **Position:** {transform.get('position', 'N/A')}")
+ state = actor.get("state", {})
+ lines.append(f"- **State Topic:** `{state.get('topic', 'N/A')}`")
+ effectors = actor.get("effectors", [])
+ if effectors:
+ lines.append("- **Effectors:**")
+ for eff in effectors:
+ lines.append(f" - **ID:** `{eff.get('id', 'N/A')}`, **State Topic:** `{eff.get('state', {}).get('topic', 'N/A')}`")
+ flight_deck = actor.get("flight_deck", [])
+ if flight_deck:
+ lines.append("- **Flight Deck:**")
+ for fd in flight_deck:
+ lines.append(f" - **ID:** `{fd.get('id', 'N/A')}`, **State Topic:** `{fd.get('state', {}).get('topic', 'N/A')}`")
+ lines.append("")
+ else:
+ lines.append("No actors defined.")
+ lines.append("")
+
+ # Sensors
+ sensors = world.get("sensors", [])
+ lines.append("## Sensors")
+ lines.append("")
+ if sensors:
+ for sensor in sensors:
+ lines.append(f"### Sensor: {sensor.get('sensor_name', 'Unnamed')}")
+ lines.append(f"- **Type:** {sensor.get('type', 'N/A')}")
+ lines.append(f"- **Parent Actor:** {sensor.get('parent', 'N/A')}")
+ transform = sensor.get("transform", {})
+ lines.append(f"- **Translation:** {transform.get('translation', 'N/A')}")
+ parameters = sensor.get("parameters", {})
+ lines.append("- **Parameters:**")
+ lines.append(f" - **Resolution:** {parameters.get('resolution', 'N/A')}")
+ lines.append(f" - **Frame Rate:** {parameters.get('frame_rate', 'N/A')} fps")
+ lines.append(f" - **Field of View:** {parameters.get('fov', 'N/A')}°")
+ lines.append(f" - **Capture Enabled:** {parameters.get('capture_enabled', 'N/A')}")
+ lines.append("")
+ else:
+ lines.append("No sensors defined.")
+ lines.append("")
+
+ # Renderers
+ renderers = scenario.get("renderers", [])
+ lines.append("## Renderers")
+ lines.append("")
+ if renderers:
+ for renderer in renderers:
+ lines.append(f"### Renderer ID: {renderer.get('renderer_id', 'N/A')}")
+ lines.append(f"- **Role:** {renderer.get('role', 'N/A')}")
+ sensors_used = renderer.get("sensors", [])
+ lines.append(f"- **Sensors:** {', '.join(sensors_used) if sensors_used else 'None'}")
+ active_camera = renderer.get("viewport_config", {}).get("active_camera", "N/A")
+ lines.append(f"- **Active Camera:** {active_camera}")
+ lines.append("")
+ else:
+ lines.append("No renderers defined.")
+ lines.append("")
+
+ # FMU Models
+ fmu_models = scenario.get("fmu_models", [])
+ lines.append("## FMU Models")
+ lines.append("")
+ if fmu_models:
+ for fmu in fmu_models:
+ lines.append(f"### FMU: {fmu.get('id', 'Unnamed')}")
+ lines.append(f"- **FMU Model Path:** {fmu.get('fmu_model_path', 'N/A')}")
+ inputs = fmu.get("component_input_topics", [])
+ outputs = fmu.get("component_output_topics", [])
+ if inputs:
+ lines.append("- **Input Topics:**")
+ for inp in inputs:
+ lines.append(f" - `{inp.get('topic', 'N/A')}` (Msg Type: {inp.get('msg_type', 'N/A')})")
+ if outputs:
+ lines.append("- **Output Topics:**")
+ for out in outputs:
+ lines.append(f" - `{out.get('topic', 'N/A')}` (Msg Type: {out.get('msg_type', 'N/A')})")
+
+ # Waypoints file if exists
+ fmu_init = fmu.get("fmu_initial_vals", {})
+ waypoints = fmu_init.get("waypoints_json_path")
+ if waypoints:
+ lines.append(f"- **Waypoints File:** `{waypoints}`")
+ lines.append("")
+ else:
+ lines.append("No FMU models defined.")
+ lines.append("")
+
+ lines.append("---")
+ lines.append("This report was auto-generated from the scenario JSON file.")
+ return "\n".join(lines)
+
+
+def save_markdown_report(md_text, filename="scenario_report.md"):
+ with open(filename, "w", encoding="utf-8") as md_file:
+ md_file.write(md_text)
+ print(f"Markdown report saved as {filename}")
+
+
+def plot_trajectories(scenario, output_filename="trajectories.png"):
+ """
+ For each FMU model that contains a 'waypoints_json_path' in its 'fmu_initial_vals',
+ read the trajectory data and plot the trajectories on a 5x5 mile map.
+ The trajectory JSON is expected to be a list of dicts with keys: 'time', 'lat', 'lon', 'alt'.
+ """
+ try:
+ import matplotlib.pyplot as plt
+ import geopandas as gpd
+ from shapely.geometry import LineString
+ import contextily as ctx
+ except ImportError:
+ print("Error: To plot trajectories, please install: geopandas, contextily, shapely, matplotlib")
+ return
+
+ fmu_models = scenario.get("fmu_models", [])
+ trajectories = []
+ labels = []
+
+ for fmu in fmu_models:
+ init_vals = fmu.get("fmu_initial_vals", {})
+ waypoints_path = init_vals.get("waypoints_json_path")
+ if waypoints_path and os.path.exists(waypoints_path):
+ with open(waypoints_path, "r", encoding="utf-8") as wf:
+ wp_data = json.load(wf)
+ # Create a list of (lon, lat) tuples from the waypoints
+ coords = [(point["lon"], point["lat"]) for point in wp_data if "lon" in point and "lat" in point]
+ if coords:
+ trajectories.append(LineString(coords))
+ labels.append(fmu.get("id", "unknown"))
+
+ if not trajectories:
+ print("No trajectories found to plot.")
+ return
+
+ # Create a GeoDataFrame using the trajectories as the geometry column
+ gdf = gpd.GeoDataFrame({"label": labels}, geometry=trajectories, crs="EPSG:4326")
+ # Convert to Web Mercator for basemap compatibility
+ gdf = gdf.to_crs(epsg=3857)
+
+ # Set a fixed square extent (5 miles x 5 miles).
+ # 5 miles is roughly 8046.72 meters.
+ square_size = 8046.72 # meters
+ half_square = square_size / 2
+
+ # Get the center of the trajectories' bounding box
+ minx, miny, maxx, maxy = gdf.total_bounds
+ center_x = (minx + maxx) / 2
+ center_y = (miny + maxy) / 2
+
+ fig, ax = plt.subplots(figsize=(10, 10))
+ # Plot each trajectory with a label
+ for idx, row in gdf.iterrows():
+ gdf.iloc[[idx]].plot(ax=ax, linewidth=2, label=row["label"])
+
+ # Set the map extent to be a square of 5 miles (8046.72 m) centered at the calculated center
+ ax.set_xlim(center_x - half_square, center_x + half_square)
+ ax.set_ylim(center_y - half_square, center_y + half_square)
+
+ ax.set_axis_off()
+ # Use OpenStreetMap's Mapnik provider as the basemap
+ ctx.add_basemap(ax, source=ctx.providers.OpenStreetMap.Mapnik)
+
+ plt.legend()
+ plt.title("Trajectories of Vehicles (5x5 Mile Map)")
+ plt.savefig(output_filename, bbox_inches="tight")
+ plt.close()
+ print(f"Trajectories plot saved as {output_filename}")
+
+
+def main():
+ # Determine the scenario JSON file from command-line (or default to scenario.json)
+ if len(sys.argv) > 1:
+ json_filename = sys.argv[1]
+ else:
+ json_filename = "test.json"
+
+ if not os.path.exists(json_filename):
+ print(f"ERROR: File '{json_filename}' not found!")
+ sys.exit(1)
+
+ with open(json_filename, "r", encoding="utf-8") as f:
+ scenario = json.load(f)
+
+ # Generate and save the Markdown report
+ report_md = generate_markdown_report(scenario)
+ save_markdown_report(report_md)
+
+ # Generate and save the trajectories visualization (fixed 5x5 mile map)
+ plot_trajectories(scenario)
+
+
+if __name__ == '__main__':
+ main()
diff --git a/aerosim/src/aerosim/utils/tracks_from_map.py b/aerosim/src/aerosim/utils/tracks_from_map.py
new file mode 100644
index 0000000..474a2b4
--- /dev/null
+++ b/aerosim/src/aerosim/utils/tracks_from_map.py
@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+"""
+This script reads a KML file containing track data (from Google Earth),
+extracts each track’s coordinates and associated metadata, and converts
+them into a CSV file (kml_output.csv) that follows the expected simulator format.
+"""
+
+import csv
+import xml.etree.ElementTree as ET
+from datetime import datetime, timedelta, timezone
+import re
+
+# Expected CSV columns
+FIELDNAMES = [
+ "eventName", "trackId", "sourceId", "trackType", "timestamp", "data",
+ "altitudeReference", "sourceTrackId", "longitude", "latitude",
+ "altitude", "aglAltitude", "mslAltitude", "wgs84Altitude"
+]
+
+def extract_track_id(name_text):
+ """
+ Extracts a numeric track ID from the Placemark name.
+ For example, if the name is "Trajectory 10000", returns 10000.
+ If no numeric ID is found, returns None.
+ """
+ match = re.search(r'\d+', name_text)
+ if match:
+ return int(match.group())
+ return None
+
+def parse_kml(kml_file):
+ """
+ Parse the KML file and return a dictionary where keys are track IDs
+ and values are lists of coordinate tuples (lon, lat, alt).
+ Assumes each Placemark contains a LineString with a
element.
+ """
+ ns = {"kml": "http://www.opengis.net/kml/2.2"}
+ tree = ET.parse(kml_file)
+ root = tree.getroot()
+
+ tracks = {}
+ for placemark in root.findall(".//kml:Placemark", ns):
+ name_elem = placemark.find("kml:name", ns)
+ if name_elem is not None:
+ name_text = name_elem.text
+ track_id = extract_track_id(name_text)
+ else:
+ track_id = None
+ # If no track_id found, assign a default value (e.g., 99999)
+ if track_id is None:
+ track_id = 99999
+
+ # Get the coordinates from the LineString element
+ coord_elem = placemark.find(".//kml:LineString/kml:coordinates", ns)
+ if coord_elem is None:
+ continue
+ coord_text = coord_elem.text.strip()
+ # Coordinates in KML are typically a space-separated list: "lon,lat,alt lon,lat,alt ..."
+ coord_list = []
+ for line in coord_text.split():
+ parts = line.split(',')
+ if len(parts) >= 3:
+ try:
+ lon = float(parts[0])
+ lat = float(parts[1])
+ alt = float(parts[2])
+ coord_list.append((lon, lat, alt))
+ except ValueError:
+ continue
+ tracks[track_id] = coord_list
+ return tracks
+
+def generate_csv_from_tracks(tracks, output_csv, interval_seconds=10):
+ """
+ Generate a CSV file (output_csv) from the given tracks dictionary.
+ Each track point is written as one row.
+ """
+ base_time = datetime.now(timezone.utc)
+ rows = []
+ for track_id, coordinates in tracks.items():
+ # Use a fixed source for KML-converted tracks
+ source_id = "GOOGLE_EARTH"
+ track_type = "Surveillance"
+ altitude_reference = "MSL"
+ # Build sourceTrackId from source and track id.
+ source_track_id = f"{source_id}::{track_id:05d}"
+ # For each point, generate a row with a timestamp that increments by interval_seconds.
+ for i, (lon, lat, alt) in enumerate(coordinates):
+ timestamp = (base_time + timedelta(seconds=i * interval_seconds)).isoformat() + "Z"
+ row = {
+ "eventName": "track.appended",
+ "trackId": f"{track_id:05d}",
+ "sourceId": source_id,
+ "trackType": track_type,
+ "timestamp": timestamp,
+ "data": "KML_TRACK", # Placeholder data field
+ "altitudeReference": altitude_reference,
+ "sourceTrackId": source_track_id,
+ "longitude": lon,
+ "latitude": lat,
+ "altitude": alt,
+ "aglAltitude": alt * 0.35, # Example computation for AGL altitude
+ "mslAltitude": alt,
+ "wgs84Altitude": alt - 50 # Example offset for WGS84 altitude
+ }
+ rows.append(row)
+ with open(output_csv, "w", newline="") as csvfile:
+ writer = csv.DictWriter(csvfile, fieldnames=FIELDNAMES)
+ writer.writeheader()
+ writer.writerows(rows)
+ print(f"KML data converted and saved to '{output_csv}'")
+
+def main():
+ import argparse
+ parser = argparse.ArgumentParser(description="Convert a KML file of tracks to CSV format for simulator use.")
+ parser.add_argument("input_kml", help="Input KML file containing track data.")
+ parser.add_argument("-o", "--output_csv", default="kml_output.csv",
+ help="Output CSV file (default: kml_output.csv)")
+ parser.add_argument("--interval", type=int, default=10,
+ help="Time interval in seconds between track points (default: 10)")
+ args = parser.parse_args()
+
+ tracks = parse_kml(args.input_kml)
+ if not tracks:
+ print("No tracks found in the KML file.")
+ return
+ generate_csv_from_tracks(tracks, args.output_csv, interval_seconds=args.interval)
+
+if __name__ == "__main__":
+ main()
diff --git a/aerosim/src/aerosim/utils/utils.py b/aerosim/src/aerosim/utils/utils.py
new file mode 100644
index 0000000..85a8b4a
--- /dev/null
+++ b/aerosim/src/aerosim/utils/utils.py
@@ -0,0 +1,307 @@
+#!/usr/bin/env python3
+"""
+Utils module for simulation tools.
+
+This module provides various utility functions including:
+- Conversion between geodetic (LLA) and ECEF coordinates.
+- Conversion between ECEF and NED coordinates.
+- LLA to NED conversion and its inverse (NED to LLA).
+- Processing CSV files with waypoints to compute NED coordinates.
+- Distance calculations using the Haversine formula and 3D Euclidean metric.
+- Bearing (forward azimuth) calculation between two points.
+"""
+
+import numpy as np
+import pandas as pd
+
+# WGS84 ellipsoid constants
+A = 6378137.0 # semi-major axis in meters
+E_SQ = 6.69437999014e-3 # first eccentricity squared
+
+def geodetic_to_ecef(lat, lon, alt):
+ """
+ Convert geodetic coordinates (latitude, longitude, altitude) to ECEF.
+
+ Parameters:
+ lat (float): Latitude in degrees.
+ lon (float): Longitude in degrees.
+ alt (float): Altitude in meters.
+
+ Returns:
+ tuple: (x, y, z) in meters.
+ """
+ lat_rad = np.radians(lat)
+ lon_rad = np.radians(lon)
+ N = A / np.sqrt(1 - E_SQ * np.sin(lat_rad)**2)
+ x = (N + alt) * np.cos(lat_rad) * np.cos(lon_rad)
+ y = (N + alt) * np.cos(lat_rad) * np.sin(lon_rad)
+ z = (N * (1 - E_SQ) + alt) * np.sin(lat_rad)
+ return x, y, z
+
+def ecef_to_geodetic(x, y, z):
+ """
+ Convert ECEF coordinates (x, y, z) to geodetic coordinates.
+
+ Parameters:
+ x (float): ECEF x-coordinate in meters.
+ y (float): ECEF y-coordinate in meters.
+ z (float): ECEF z-coordinate in meters.
+
+ Returns:
+ tuple: (latitude in degrees, longitude in degrees, altitude in meters).
+ """
+ lon = np.arctan2(y, x)
+ p = np.sqrt(x*x + y*y)
+ lat = np.arctan2(z, p * (1 - E_SQ)) # initial guess
+ alt = 0
+ # Iterative improvement (5 iterations usually suffice)
+ for _ in range(5):
+ N = A / np.sqrt(1 - E_SQ * np.sin(lat)**2)
+ alt = p / np.cos(lat) - N
+ lat = np.arctan2(z + E_SQ * N * np.sin(lat), p)
+ return np.degrees(lat), np.degrees(lon), alt
+
+def ecef_to_ned(dx, dy, dz, ref_lat, ref_lon):
+ """
+ Convert differences in ECEF coordinates (dx, dy, dz) to NED coordinates.
+
+ Parameters:
+ dx, dy, dz (float): Differences in ECEF coordinates (meters).
+ ref_lat (float): Reference latitude in degrees.
+ ref_lon (float): Reference longitude in degrees.
+
+ Returns:
+ tuple: (north, east, down) in meters.
+ """
+ lat_rad = np.radians(ref_lat)
+ lon_rad = np.radians(ref_lon)
+ north = -np.sin(lat_rad)*np.cos(lon_rad)*dx - np.sin(lat_rad)*np.sin(lon_rad)*dy + np.cos(lat_rad)*dz
+ east = -np.sin(lon_rad)*dx + np.cos(lon_rad)*dy
+ down = -np.cos(lat_rad)*np.cos(lon_rad)*dx - np.cos(lat_rad)*np.sin(lon_rad)*dy - np.sin(lat_rad)*dz
+ return north, east, down
+
+def ned_to_ecef(north, east, down, ref_lat, ref_lon):
+ """
+ Convert NED coordinates (north, east, down) to differences in ECEF coordinates (dx, dy, dz).
+
+ Parameters:
+ north, east, down (float): NED coordinates in meters.
+ ref_lat (float): Reference latitude in degrees.
+ ref_lon (float): Reference longitude in degrees.
+
+ Returns:
+ tuple: (dx, dy, dz) in meters.
+ """
+ lat_rad = np.radians(ref_lat)
+ lon_rad = np.radians(ref_lon)
+ # Inverse of the rotation matrix (transpose) is used here.
+ dx = -np.sin(lat_rad)*np.cos(lon_rad)*north - np.sin(lon_rad)*east - np.cos(lat_rad)*np.cos(lon_rad)*down
+ dy = -np.sin(lat_rad)*np.sin(lon_rad)*north + np.cos(lon_rad)*east - np.cos(lat_rad)*np.sin(lon_rad)*down
+ dz = np.cos(lat_rad)*north - np.sin(lat_rad)*down
+ return dx, dy, dz
+
+def lla_to_ned(lat, lon, alt, ref_lat, ref_lon, ref_alt):
+ """
+ Convert a single LLA point to NED coordinates relative to a reference LLA point.
+
+ Parameters:
+ lat, lon, alt (float): Target point in LLA (degrees, degrees, meters).
+ ref_lat, ref_lon, ref_alt (float): Reference point in LLA (degrees, degrees, meters).
+
+ Returns:
+ tuple: (north, east, down) in meters.
+ """
+ target_x, target_y, target_z = geodetic_to_ecef(lat, lon, alt)
+ ref_x, ref_y, ref_z = geodetic_to_ecef(ref_lat, ref_lon, ref_alt)
+ dx = target_x - ref_x
+ dy = target_y - ref_y
+ dz = target_z - ref_z
+ return ecef_to_ned(dx, dy, dz, ref_lat, ref_lon)
+
+def ned_to_lla(north, east, down, ref_lat, ref_lon, ref_alt):
+ """
+ Convert a single NED point relative to a reference LLA point back to LLA coordinates.
+
+ Parameters:
+ north, east, down (float): NED coordinates in meters.
+ ref_lat, ref_lon, ref_alt (float): Reference LLA point (degrees, degrees, meters).
+
+ Returns:
+ tuple: (latitude in degrees, longitude in degrees, altitude in meters).
+ """
+ ref_x, ref_y, ref_z = geodetic_to_ecef(ref_lat, ref_lon, ref_alt)
+ dx, dy, dz = ned_to_ecef(north, east, down, ref_lat, ref_lon)
+ x = ref_x + dx
+ y = ref_y + dy
+ z = ref_z + dz
+ return ecef_to_geodetic(x, y, z)
+
+def process_trajectory(input_file, output_file, ref_lat=None, ref_lon=None, ref_alt=None):
+ """
+ Process a CSV file of LLA waypoints, compute and append NED coordinates, then save the result.
+
+ If no reference point is provided, the first waypoint in the CSV is used.
+
+ Parameters:
+ input_file (str): Path to the input CSV file.
+ output_file (str): Path to save the output CSV file.
+ ref_lat, ref_lon, ref_alt (float, optional): Reference LLA point.
+ """
+ try:
+ df = pd.read_csv(input_file)
+ except Exception as e:
+ print(f"Error reading CSV file: {e}")
+ return
+
+ cols_lower = [str(col).strip().lower() for col in df.columns]
+ lat_idx, lon_idx, alt_idx = None, None, None
+ for i, col in enumerate(cols_lower):
+ if "lat" in col and lat_idx is None:
+ lat_idx = i
+ elif ("lon" in col or "long" in col) and lon_idx is None:
+ lon_idx = i
+ elif "alt" in col and alt_idx is None:
+ alt_idx = i
+ if lat_idx is None or lon_idx is None or alt_idx is None:
+ lat_series = df[df.columns[0]]
+ lon_series = df[df.columns[1]]
+ alt_series = df[df.columns[2]]
+ else:
+ lat_series = df[df.columns[lat_idx]]
+ lon_series = df[df.columns[lon_idx]]
+ alt_series = df[df.columns[alt_idx]]
+
+ if ref_lat is None or ref_lon is None or ref_alt is None:
+ ref_lat = lat_series.iloc[0]
+ ref_lon = lon_series.iloc[0]
+ ref_alt = alt_series.iloc[0]
+ print(f"No reference point provided. Using first waypoint as reference: ({ref_lat}, {ref_lon}, {ref_alt})")
+
+ ref_x, ref_y, ref_z = geodetic_to_ecef(ref_lat, ref_lon, ref_alt)
+ north_list, east_list, down_list = [], [], []
+
+ for i in range(len(df)):
+ lat = lat_series.iloc[i]
+ lon = lon_series.iloc[i]
+ alt = alt_series.iloc[i]
+ x, y, z = geodetic_to_ecef(lat, lon, alt)
+ dx = x - ref_x
+ dy = y - ref_y
+ dz = z - ref_z
+ north, east, down = ecef_to_ned(dx, dy, dz, ref_lat, ref_lon)
+ north_list.append(north)
+ east_list.append(east)
+ down_list.append(down)
+
+ df['north'] = north_list
+ df['east'] = east_list
+ df['down'] = down_list
+
+ if output_file is None:
+ output_file = "ned_output.csv"
+
+ df.to_csv(output_file, index=False)
+ print(f"NED coordinates saved to {output_file}")
+
+def process_single_point(lat, lon, alt, ref_lat, ref_lon, ref_alt, output_file=None):
+ """
+ Process a single LLA point to compute its NED coordinates relative to a reference point.
+
+ Parameters:
+ lat, lon, alt (float): Target point in LLA (degrees, degrees, meters).
+ ref_lat, ref_lon, ref_alt (float): Reference LLA point (degrees, degrees, meters).
+ output_file (str, optional): If provided, the result is saved to this CSV file.
+ """
+ north, east, down = lla_to_ned(lat, lon, alt, ref_lat, ref_lon, ref_alt)
+ print("Single point conversion:")
+ print(f"Input LLA: ({lat}, {lon}, {alt})")
+ print(f"Reference LLA: ({ref_lat}, {ref_lon}, {ref_alt})")
+ print(f"NED: North = {north:.3f} m, East = {east:.3f} m, Down = {down:.3f} m")
+
+ if output_file is not None:
+ data = {
+ "lat": [lat],
+ "lon": [lon],
+ "alt": [alt],
+ "north": [north],
+ "east": [east],
+ "down": [down]
+ }
+ result_df = pd.DataFrame(data)
+ result_df.to_csv(output_file, index=False)
+ print(f"Result saved to {output_file}")
+
+def haversine_distance(lat1, lon1, lat2, lon2):
+ """
+ Calculate the great-circle distance between two points on the Earth's surface using the Haversine formula.
+
+ Parameters:
+ lat1, lon1 (float): Latitude and longitude of the first point (in degrees).
+ lat2, lon2 (float): Latitude and longitude of the second point (in degrees).
+
+ Returns:
+ float: Distance in meters.
+ """
+ R = A # Using the WGS84 semi-major axis as an approximation for Earth's radius.
+ lat1_rad, lon1_rad = np.radians(lat1), np.radians(lon1)
+ lat2_rad, lon2_rad = np.radians(lat2), np.radians(lon2)
+ dlat = lat2_rad - lat1_rad
+ dlon = lon2_rad - lon1_rad
+ a_val = np.sin(dlat/2)**2 + np.cos(lat1_rad) * np.cos(lat2_rad) * np.sin(dlon/2)**2
+ c_val = 2 * np.arctan2(np.sqrt(a_val), np.sqrt(1 - a_val))
+ distance = R * c_val
+ return distance
+
+def euclidean_distance_lla(lat1, lon1, alt1, lat2, lon2, alt2):
+ """
+ Calculate the 3D Euclidean distance between two LLA points by converting them to ECEF.
+
+ Parameters:
+ lat1, lon1, alt1 (float): First point (degrees, degrees, meters).
+ lat2, lon2, alt2 (float): Second point (degrees, degrees, meters).
+
+ Returns:
+ float: Distance in meters.
+ """
+ x1, y1, z1 = geodetic_to_ecef(lat1, lon1, alt1)
+ x2, y2, z2 = geodetic_to_ecef(lat2, lon2, alt2)
+ return np.sqrt((x2 - x1)**2 + (y2 - y1)**2 + (z2 - z1)**2)
+
+def bearing_between_points(lat1, lon1, lat2, lon2):
+ """
+ Calculate the initial bearing (forward azimuth) from point 1 to point 2.
+
+ Parameters:
+ lat1, lon1 (float): Latitude and longitude of the first point (in degrees).
+ lat2, lon2 (float): Latitude and longitude of the second point (in degrees).
+
+ Returns:
+ float: Bearing in degrees (0° = north).
+ """
+ lat1_rad, lon1_rad = np.radians(lat1), np.radians(lon1)
+ lat2_rad, lon2_rad = np.radians(lat2), np.radians(lon2)
+ dlon = lon2_rad - lon1_rad
+ x = np.sin(dlon) * np.cos(lat2_rad)
+ y = np.cos(lat1_rad) * np.sin(lat2_rad) - np.sin(lat1_rad) * np.cos(lat2_rad) * np.cos(dlon)
+ bearing_rad = np.arctan2(x, y)
+ bearing_deg = (np.degrees(bearing_rad) + 360) % 360
+ return bearing_deg
+
+if __name__ == "__main__":
+ # Example usage: LLA to NED conversion
+ lat, lon, alt = 37.7749, -122.4194, 10 # Example: San Francisco (approx.)
+ ref_lat, ref_lon, ref_alt = 37.7749, -122.4194, 0 # Using same lat/lon, different altitude as reference
+ print("LLA to NED conversion example:")
+ north, east, down = lla_to_ned(lat, lon, alt, ref_lat, ref_lon, ref_alt)
+ print(f"NED: North = {north:.3f} m, East = {east:.3f} m, Down = {down:.3f} m")
+
+ # Example usage: Distance calculations between San Francisco and Los Angeles
+ lat2, lon2, alt2 = 34.0522, -118.2437, 15 # Example: Los Angeles (approx.)
+ haversine_dist = haversine_distance(lat, lon, lat2, lon2)
+ euclidean_dist = euclidean_distance_lla(lat, lon, alt, lat2, lon2, alt2)
+ print(f"Haversine distance: {haversine_dist/1000:.3f} km")
+ print(f"3D Euclidean distance: {euclidean_dist/1000:.3f} km")
+
+ # Example usage: Bearing calculation
+ bearing = bearing_between_points(lat, lon, lat2, lon2)
+ print(f"Initial bearing from San Francisco to Los Angeles: {bearing:.3f}°")
diff --git a/aerosim/src/aerosim/utils/visualization.py b/aerosim/src/aerosim/utils/visualization.py
new file mode 100644
index 0000000..23d5688
--- /dev/null
+++ b/aerosim/src/aerosim/utils/visualization.py
@@ -0,0 +1,491 @@
+import os
+import csv
+import json
+import random
+from datetime import datetime
+
+import pandas as pd
+import folium
+import matplotlib.colors as mcolors
+
+###############################
+# Data Loading and Normalization
+###############################
+
+def load_data_to_df(input_file, default_track_id=None):
+ """
+ Load data from a CSV or JSON file into a pandas DataFrame.
+
+ For JSON files, if the keys "lat", "lon", "alt", and "time" exist,
+ they are renamed to "latitude", "longitude", "altitude", and "timestamp"
+ to match the CSV schema.
+
+ If the "trackId" column is missing, a default value is assigned.
+ By default, this value is taken as the name of the folder containing the file,
+ or "trajectory" if no folder exists.
+
+ Args:
+ input_file (str): Path to the CSV or JSON file.
+ default_track_id (str, optional): Default track ID to use if missing.
+
+ Returns:
+ pd.DataFrame: DataFrame with normalized column names.
+ """
+ _, ext = os.path.splitext(input_file)
+ ext = ext.lower()
+ if ext == '.csv':
+ df = pd.read_csv(input_file)
+ elif ext == '.json':
+ try:
+ df = pd.read_json(input_file)
+ except ValueError:
+ with open(input_file, 'r', encoding='utf-8') as f:
+ data = json.load(f)
+ df = pd.DataFrame(data)
+ else:
+ raise ValueError("Unsupported file format. Only CSV and JSON are supported.")
+
+ # Normalize column names for JSON trajectory files.
+ if 'lat' in df.columns and 'lon' in df.columns:
+ df.rename(columns={'lat': 'latitude', 'lon': 'longitude'}, inplace=True)
+ if 'alt' in df.columns:
+ df.rename(columns={'alt': 'altitude'}, inplace=True)
+ if 'time' in df.columns:
+ df.rename(columns={'time': 'timestamp'}, inplace=True)
+
+ # Determine default_track_id from the folder name if not provided.
+ if default_track_id is None:
+ folder = os.path.dirname(input_file)
+ if folder:
+ default_track_id = os.path.basename(folder)
+ else:
+ default_track_id = "trajectory"
+
+ # If there is no "trackId" column, add one using the default track ID.
+ if 'trackId' not in df.columns:
+ df['trackId'] = default_track_id
+
+ return df
+
+def combine_data_from_folder(input_folder):
+ """
+ Load and combine data from all CSV/JSON files in the specified folder into a single DataFrame.
+
+ Args:
+ input_folder (str): Directory containing CSV/JSON files.
+
+ Returns:
+ pd.DataFrame: Combined DataFrame (or an empty DataFrame if none found).
+ """
+ all_dfs = []
+ for filename in os.listdir(input_folder):
+ full_path = os.path.join(input_folder, filename)
+ if os.path.isfile(full_path) and os.path.splitext(filename)[1].lower() in ['.csv', '.json']:
+ try:
+ df = load_data_to_df(full_path)
+ all_dfs.append(df)
+ except Exception as e:
+ print(f"Error loading {filename}: {e}")
+ if all_dfs:
+ combined_df = pd.concat(all_dfs, ignore_index=True)
+ return combined_df
+ else:
+ return pd.DataFrame()
+
+###############################
+# Folium-based Visualization #
+###############################
+
+def generate_tooltip(row):
+ """
+ Generate an HTML tooltip from a DataFrame row.
+
+ Uses a detailed tooltip if expected keys (e.g., "eventName") exist;
+ otherwise, produces a simplified version.
+ """
+ if 'eventName' in row:
+ tooltip = (
+ f"Event Name: {row['eventName']}
"
+ f"Track ID: {row['trackId']}
"
+ f"Source ID: {row['sourceId']}
"
+ f"Track Type: {row['trackType']}
"
+ f"Timestamp: {row['timestamp']}
"
+ f"Data: {row['data']}
"
+ f"Altitude Reference: {row['altitudeReference']}
"
+ f"Source Track ID: {row['sourceTrackId']}
"
+ f"Longitude: {row['longitude']}
"
+ f"Latitude: {row['latitude']}
"
+ f"Altitude: {row['altitude']}
"
+ f"AGL Altitude: {row['aglAltitude']}
"
+ f"MSL Altitude: {row['mslAltitude']}
"
+ f"WGS84 Altitude: {row['wgs84Altitude']}"
+ )
+ else:
+ tooltip = (
+ f"Timestamp: {row['timestamp']}
"
+ f"Altitude: {row.get('altitude', '')}"
+ )
+ return tooltip
+
+def plot_tracks(input_file, output_html):
+ """
+ Plot track data from a single CSV or JSON file on an interactive folium map with tooltips.
+
+ Args:
+ input_file (str): Path to the CSV or JSON file.
+ output_html (str): Path to save the generated HTML map.
+ """
+ df = load_data_to_df(input_file)
+ _plot_df(df, output_html)
+
+def _plot_df(df, output_html):
+ """
+ Internal helper to plot a DataFrame of track data onto a folium map.
+
+ Args:
+ df (pd.DataFrame): DataFrame containing track data.
+ output_html (str): Path to save the generated HTML map.
+ """
+ if df.empty:
+ print("No data to plot.")
+ return
+ center_lat = df["latitude"].mean()
+ center_lon = df["longitude"].mean()
+ m = folium.Map(location=[center_lat, center_lon], zoom_start=10)
+
+ track_ids = df["trackId"].unique()
+ color_list = list(mcolors.CSS4_COLORS.values())
+ random.shuffle(color_list)
+
+ for i, track_id in enumerate(track_ids):
+ track_data = df[df["trackId"] == track_id].sort_values("timestamp")
+ points = list(zip(track_data["latitude"], track_data["longitude"]))
+ color = color_list[i % len(color_list)]
+ if len(points) > 1:
+ folium.PolyLine(points, color=color, weight=2.5, opacity=1).add_to(m)
+ for idx, row in track_data.iterrows():
+ tooltip_text = generate_tooltip(row)
+ folium.CircleMarker(
+ location=(row["latitude"], row["longitude"]),
+ radius=4,
+ color=color,
+ fill=True,
+ fill_color=color,
+ fill_opacity=0.7,
+ tooltip=tooltip_text
+ ).add_to(m)
+ m.save(output_html)
+ print(f"Combined map with tooltips has been saved to {output_html}")
+
+def plot_combined_tracks_from_folder(input_folder, output_html):
+ """
+ Combine all CSV/JSON files in a folder and plot the tracks on a single folium map.
+
+ Args:
+ input_folder (str): Directory containing CSV/JSON files.
+ output_html (str): Path to save the generated HTML map.
+ """
+ combined_df = combine_data_from_folder(input_folder)
+ if combined_df.empty:
+ print("No valid data found in folder.")
+ return
+ _plot_df(combined_df, output_html)
+
+###############################
+# KML-based Visualization #
+###############################
+
+def parse_csv(file_path):
+ """
+ Parse a CSV file and return a list of dictionaries.
+ """
+ with open(file_path, newline='', encoding='utf-8') as f:
+ reader = csv.DictReader(f)
+ return list(reader)
+
+def parse_json(file_path):
+ """
+ Parse a JSON file and return its content.
+ """
+ with open(file_path, 'r', encoding='utf-8') as f:
+ return json.load(f)
+
+def unify_records(records, input_format):
+ """
+ Convert records into a unified format with keys:
+ - track_id, timestamp, longitude, latitude, altitude
+ """
+ unified = []
+ for idx, rec in enumerate(records):
+ if input_format == 'csv':
+ if 'timestamp' in rec and 'trackId' in rec and 'wgs84Altitude' in rec:
+ track_id = rec.get('trackId', f'Point{idx}')
+ timestamp = rec.get('timestamp', '')
+ longitude = rec.get('longitude', '')
+ latitude = rec.get('latitude', '')
+ altitude = rec.get('wgs84Altitude', '')
+ else:
+ track_id = rec.get('trackId', f'Point{idx}')
+ timestamp = rec.get('timestamp', '')
+ longitude = rec.get('longitude', rec.get('lon', ''))
+ latitude = rec.get('latitude', rec.get('lat', ''))
+ altitude = rec.get('wgs84Altitude', rec.get('altitude', rec.get('alt', '0')))
+ elif input_format == 'json':
+ track_id = rec.get('trackId', 'trajectory')
+ timestamp = rec.get('time', '')
+ longitude = rec.get('lon', '')
+ latitude = rec.get('lat', '')
+ altitude = rec.get('alt', '')
+ else:
+ continue
+
+ try:
+ timestamp_val = float(timestamp)
+ except:
+ timestamp_val = timestamp
+ try:
+ longitude_val = float(longitude)
+ except:
+ longitude_val = 0.0
+ try:
+ latitude_val = float(latitude)
+ except:
+ latitude_val = 0.0
+ try:
+ altitude_val = float(altitude)
+ except:
+ altitude_val = 0.0
+
+ unified.append({
+ 'track_id': track_id,
+ 'timestamp': timestamp_val,
+ 'longitude': longitude_val,
+ 'latitude': latitude_val,
+ 'altitude': altitude_val
+ })
+ return unified
+
+def unify_dataframe(df):
+ """
+ Convert a DataFrame into a list of unified records with keys:
+ - track_id, timestamp, longitude, latitude, altitude
+ """
+ records = []
+ for idx, row in df.iterrows():
+ try:
+ timestamp_val = float(row['timestamp'])
+ except:
+ timestamp_val = row['timestamp']
+ try:
+ longitude_val = float(row['longitude'])
+ except:
+ longitude_val = 0.0
+ try:
+ latitude_val = float(row['latitude'])
+ except:
+ latitude_val = 0.0
+ try:
+ altitude_val = float(row['altitude'])
+ except:
+ altitude_val = 0.0
+ record = {
+ 'track_id': row['trackId'],
+ 'timestamp': timestamp_val,
+ 'longitude': longitude_val,
+ 'latitude': latitude_val,
+ 'altitude': altitude_val
+ }
+ records.append(record)
+ return records
+
+def group_by_track(unified_records):
+ groups = {}
+ for rec in unified_records:
+ tid = rec['track_id']
+ groups.setdefault(tid, []).append(rec)
+ for tid in groups:
+ groups[tid] = sorted(groups[tid], key=lambda x: x['timestamp'])
+ return groups
+
+def get_color(index):
+
+ colors = [
+ "ff0000ff", # Red
+ "ff00ff00", # Green
+ "ffff0000", # Blue
+ "ff00ffff", # Yellow
+ "ffffff00", # Cyan
+ "ff990099", # Purple
+ "ff0099ff", # Orange-ish
+ "ff99ff00", # Lime
+ ]
+ return colors[index % len(colors)]
+
+
+def generate_styles(track_ids):
+ styles = []
+ for idx, tid in enumerate(sorted(track_ids)):
+ color = get_color(idx)
+ style = f"""
+
+ """.strip()
+ styles.append(style)
+ return styles
+
+def generate_trajectory_placemarks(groups):
+ placemarks = []
+ for tid, points in groups.items():
+ coords = " ".join([f"{pt['longitude']},{pt['latitude']},{pt['altitude']}" for pt in points])
+ placemark = f"""
+
+ {tid}
+ #style_{tid}
+
+ 1
+ 1
+ absolute
+
+ {coords}
+
+
+
+ """.strip()
+ placemarks.append(placemark)
+ return placemarks
+
+def generate_marker_placemarks(groups):
+ markers = []
+ for tid, points in groups.items():
+ for pt in points:
+ placemark = f"""
+
+ {tid}
+ #style_{tid}
+ Timestamp: {pt['timestamp']}
+
+ {pt['longitude']},{pt['latitude']},{pt['altitude']}
+
+
+ """.strip()
+ markers.append(placemark)
+ return markers
+
+def create_kml(styles, trajectory_placemarks, marker_placemarks):
+ kml_content = f"""
+
+
+ 3D Trajectories with Markers
+ {''.join(styles)}
+ {''.join(trajectory_placemarks)}
+ {''.join(marker_placemarks)}
+
+"""
+ return kml_content
+
+def convert_waypoints_to_kml(input_file, output_file=None, input_format=None):
+ if input_format is None:
+ _, ext = os.path.splitext(input_file)
+ ext = ext.lower()
+ if ext == '.json':
+ input_format = 'json'
+ elif ext == '.csv':
+ input_format = 'csv'
+ else:
+ raise ValueError("Could not determine input format. Please specify input_format='csv' or 'json'.")
+
+ if input_format == 'csv':
+ records = parse_csv(input_file)
+ elif input_format == 'json':
+ records = parse_json(input_file)
+ else:
+ raise ValueError("Invalid input_format specified.")
+
+ unified_records = unify_records(records, input_format)
+ groups = group_by_track(unified_records)
+ styles = generate_styles(groups.keys())
+ trajectory_placemarks = generate_trajectory_placemarks(groups)
+ marker_placemarks = generate_marker_placemarks(groups)
+ kml_content = create_kml(styles, trajectory_placemarks, marker_placemarks)
+
+ if output_file is None:
+ base, _ = os.path.splitext(input_file)
+ output_file = base + '.kml'
+
+ with open(output_file, 'w', encoding='utf-8') as f:
+ f.write(kml_content)
+ print(f"KML file successfully created at: {output_file}")
+
+def convert_combined_tracks_to_kml(input_folder, output_file):
+ """
+ Combine all CSV/JSON files in a folder and generate a single KML file.
+
+ Args:
+ input_folder (str): Directory containing CSV/JSON files.
+ output_file (str): Path to save the combined KML file.
+ """
+ combined_df = combine_data_from_folder(input_folder)
+ if combined_df.empty:
+ print("No valid data found in folder.")
+ return
+ records = unify_dataframe(combined_df)
+ groups = group_by_track(records)
+ styles = generate_styles(groups.keys())
+ trajectory_placemarks = generate_trajectory_placemarks(groups)
+ marker_placemarks = generate_marker_placemarks(groups)
+ kml_content = create_kml(styles, trajectory_placemarks, marker_placemarks)
+ with open(output_file, 'w', encoding='utf-8') as f:
+ f.write(kml_content)
+ print(f"Combined KML file successfully created at: {output_file}")
+
+###############################
+# Folder-Level Visualization (Combined)
+###############################
+
+def visualize_folder(input_folder, output_file, method='folium'):
+ """
+ Combine all files in a folder and generate a single visualization file.
+
+ Args:
+ input_folder (str): Directory containing input files (CSV/JSON).
+ output_file (str): Output file path (HTML for folium, KML for KML).
+ method (str): 'folium' or 'kml'. Defaults to 'folium'.
+ """
+ if method == 'folium':
+ plot_combined_tracks_from_folder(input_folder, output_file)
+ elif method == 'kml':
+ convert_combined_tracks_to_kml(input_folder, output_file)
+ else:
+ print(f"Unsupported visualization method: {method}")
+
+if __name__ == "__main__":
+ # Example usage:
+
+ # 1. Visualize a single CSV or JSON file using folium.
+ # plot_tracks("output.csv", "map_output.html")
+ # plot_tracks("trajectories.json", "trajectory_map.html")
+
+ # 2. Convert a single CSV (or JSON) file to KML.
+ # convert_waypoints_to_kml("filteredoutput.csv", "filteredoutput.kml")
+
+ # 3. Visualize all files in a folder as a combined map.
+ # For folium (single HTML map):
+ # visualize_folder("tracks", "combined_tracks_map.html", method='folium')
+ # For KML (single KML file):
+ # visualize_folder("trajectories", "combined_trajectories.kml", method='kml')
+
+ pass
diff --git a/docs/utils.md b/docs/utils.md
new file mode 100644
index 0000000..80e4781
--- /dev/null
+++ b/docs/utils.md
@@ -0,0 +1,391 @@
+# 🛩️ Aerosim Utils
+
+
+
+[](https://www.python.org/downloads/)
+
+
+A suite of tools for aircraft trajectory and scenario generation, supporting both real ADS-B data processing and artificial track generation.
+
+[Installation](#-installation) • [Quick Start](#-quick-start)
+
+
+
+## 🌟 Features
+
+
+
+| Category | Features |
+|----------|----------|
+| 📊 **OpenADSB Track Generator** | • ADS‑B JSON to CSV conversion
• Geographic filtering
• Time range filtering
• Relative ownship path filtering
• Trajectory JSON conversion
• Simulation scenario creation |
+| 🎮 **Artificial Track Generator** | • Random track generation
• Circular/elliptical patterns
• Flyby/square/rectangle patterns
• Zigzag and spiral patterns
• Ownship-relative patterns
• KML data conversion |
+| 🗺️ **Google Earth Integration** | • KML to CSV conversion
• Real-world track integration
• Custom track creation |
+| 📈 **Visualization & Conversion** | • Interactive Folium maps
• Combined KML generation
• Real-time visualization |
+| ⚙️ **Scenario Generation** | • JSON scenario creation
• Actor configuration
• Sensor setup
• FMU model integration |
+| 📝 **Scenario Reports** | • Markdown report generation
• Trajectory visualization |
+| 🛠️ **Utilities** | • LLA/NED coordinate conversion
• Distance & bearing calculations
• Interactive track visualization |
+
+
+
+## 🚀 Quick Start
+
+### Installation
+
+```bash
+# Install from PyPI (Recommended)
+pip install aerosim-utils
+
+# Or install from source
+cd aerosim-utils
+pip install -e .
+```
+
+### Basic Usage
+
+
+Process ADS-B Data
+
+```bash
+# Convert ADS-B JSON to CSV
+aerosim adsb json2csv -i inputs/ssedata.json -o outputs/output.csv
+
+# Filter tracks around a reference point
+aerosim adsb filter -i outputs/output.csv \
+ --lat 34.217411 \
+ --lon -118.491081 \
+ --radius 50 \
+ -o outputs/filtered_tracks.csv \
+ --tracks-dir tracks
+
+# Convert filtered tracks to trajectory JSON files
+aerosim adsb tracks2json -i tracks -o trajectories
+```
+
+
+
+Generate Artificial Tracks
+
+```bash
+# Generate artificial tracks using a specified maneuver
+aerosim artificial generate-tracks \
+ --num-tracks 5 \
+ --maneuver circular \
+ --center-lat 34.217411 \
+ --center-lon -118.491081 \
+ --center-alt 1000 \
+ --separation 0.005 \
+ --time-delay 30 \
+ --num-points 20 \
+ --interval 10 \
+ -o outputs/artificial_tracks.csv
+
+# Fly-along mode using an external ownship trajectory
+aerosim artificial generate-tracks \
+ --num-tracks 3 \
+ --maneuver zigzag \
+ --center-lat 33.75 \
+ --center-lon -118.25 \
+ --center-alt 1000 \
+ --separation 0.005 \
+ --time-delay 30 \
+ --num-points 20 \
+ --interval 10 \
+ -o outputs/artificial_tracks_flyalong.csv \
+ --fly-along \
+ --ownship-file inputs/ownship_trajectory.json \
+ --ownship-format json
+```
+
+
+
+Convert KML from Google Earth
+
+```bash
+# Convert a KML file from Google Earth to CSV format
+aerosim tracks kml2csv -i my_track.kml -o kml_output.csv --interval 10
+```
+
+
+
+Generate Scenarios
+
+```bash
+# Generate simulation scenario JSON from trajectory files
+aerosim scenario generate -t trajectories -o auto_gen_scenario.json
+```
+
+
+
+Visualization
+
+```bash
+# Generate a combined interactive Folium map
+aerosim vis visualize -i tracks -o visualization/folium/combined_map.html --method folium
+
+# Generate a combined KML file
+aerosim vis visualize -i trajectories -o visualization/kml/combined_map.kml --method kml
+```
+
+
+
+Scenario Reports
+
+```bash
+# Generate a markdown report from a scenario JSON file
+aerosim report scenario -i scenario.json -o reports/scenario_report.md
+
+# Generate a report with trajectory plot
+aerosim report scenario -i scenario.json -o reports/scenario_report.md --plot --plot-output reports/trajectories.png
+
+# Generate reports as part of workflow commands (enabled by default)
+aerosim workflow openadsb --input-json data.json --no-report # Disable report generation
+aerosim workflow artificial --no-plot # Disable trajectory plot
+aerosim workflow kml --report-output custom_report.md # Custom report path
+```
+
+The scenario report includes:
+- Overview of the scenario
+- Clock settings
+- Orchestrator configuration
+- World settings
+- Actor details
+- Sensor configurations
+- Renderer information
+- FMU model details
+- Optional trajectory visualization
+
+
+
+Helper Utilities
+
+```bash
+# Clamp a value between minimum and maximum
+aerosim helpers clamp --value 150 --min 0 --max 100 # Returns 100
+
+# Normalize a heading to 0-360 degrees
+aerosim helpers normalize-heading --heading 370 # Returns 10
+
+# Calculate distance and bearing between two points
+aerosim helpers distance-bearing \
+ --lat1 34.217411 \
+ --lon1 -118.491081 \
+ --lat2 34.217411 \
+ --lon2 -118.491081
+```
+
+
+## 💻 Python API Usage
+
+```python
+from aerosim_utils import ArtificialTrackGenerator, convert_json_to_csv, visualize_folder
+
+# Generate artificial tracks
+generator = ArtificialTrackGenerator(
+ num_tracks=5,
+ track_generator_class='circular',
+ center_lat=34.217411,
+ center_lon=-118.491081,
+ center_alt=1000,
+ separation_distance=0.005,
+ time_delay=30,
+ num_points=20,
+ interval_seconds=10
+)
+generator.generate()
+generator.save_to_csv("outputs/artificial_tracks.csv")
+
+# Convert ADS-B JSON data to CSV
+convert_json_to_csv("inputs/ssedata.json", "outputs/output.csv")
+
+# Visualize track data
+visualize_folder("tracks", "visualization/folium/combined_map.html", method='folium')
+```
+
+## ⚙️ Configuration
+
+Customize the package behavior using a configuration file:
+
+```bash
+# Show current configuration
+aerosim config show
+
+# Set a specific configuration value
+aerosim config set DEFAULT_CENTER_LAT 34.217411
+
+# Reset to defaults
+aerosim config reset
+
+# Validate configuration
+aerosim config validate
+```
+
+### Available Settings
+```json
+{
+ "DEFAULT_INPUT_DIR": "inputs",
+ "DEFAULT_OUTPUT_DIR": "outputs",
+ "DEFAULT_TRACKS_DIR": "tracks",
+ "DEFAULT_TRAJECTORIES_DIR": "trajectories",
+ "DEFAULT_CENTER_LAT": 34.217411,
+ "DEFAULT_CENTER_LON": -118.491081,
+ "DEFAULT_RADIUS_KM": 50.0,
+ "DEFAULT_ALTITUDE": 1000.0,
+ "DEFAULT_INTERVAL_SECONDS": 5.0,
+ "DEFAULT_SCENARIO_NAME": "auto_gen_scenario",
+ "DEFAULT_WORLD_NAME": "default_world",
+ "DEFAULT_SENSOR_CONFIG": {
+ "type": "camera",
+ "fov": 90,
+ "resolution": [1920, 1080],
+ "update_rate": 30
+ }
+}
+```
+
+
+## 🔍 Data Sources
+
+### OpenSky Network
+
+
+
+[](https://opensky-network.org/)
+
+
+
+The OpenSky Network provides open access to real‑time and historical ADS‑B data:
+
+- Registration: Sign up for a free account
+- Data Access: Use their REST API for real‑time or historical data
+- Usage: Save JSON data in the `inputs/` folder
+
+**Resources:**
+- [OpenSky Network Documentation](https://opensky-network.org/)
+- [MIT LL - em-download-opensky](https://github.com/mit-ll/em-download-opensky)
+
+### ADS-B Exchange
+
+
+
+[](https://www.adsbexchange.com/)
+
+
+
+ADSBExchange offers ADS‑B data services:
+
+- Free Feed: Available for non‑commercial projects
+- API Access: Follow their documentation and guidelines
+- Integration: Store data in `inputs/` directory
+
+## 📝 Creating Tracks in Google Earth
+
+
+Step-by-Step Guide
+
+### Step 1: Install and Open Google Earth Pro
+1. Download from [Google Earth website](https://earth.google.com/web/)
+2. Follow installation instructions
+3. Launch Google Earth Pro
+
+### Step 2: Create a Track
+1. Navigate to your area of interest
+2. Click "Add Path" icon
+3. Name your track (e.g., "Trajectory 10000")
+4. Create track points by clicking
+5. Adjust points as needed
+6. Customize appearance
+7. Click OK to save
+
+### Step 3: Export as KML
+1. Find path in "Places" panel
+2. Right-click → "Save Place As..."
+3. Choose location and filename
+4. Select KML format (not KMZ)
+5. Click Save
+
+### Step 4: Process with Aerosim Utils
+```bash
+# Convert KML to CSV
+aerosim tracks kml2csv -i my_track.kml -o kml_output.csv --interval 10
+
+# Generate scenario
+aerosim scenario generate -t trajectories -o auto_gen_scenario.json
+```
+
+
+## 🔄 Complete Processing Workflows
+
+
+
+| Workflow | Description | Command |
+|----------|-------------|---------|
+| OpenADSB | Process real ADS‑B data | `aerosim workflow openadsb` |
+| Artificial | Generate artificial tracks | `aerosim workflow artificial` |
+| KML | Process Google Earth tracks | `aerosim workflow kml` |
+
+
+
+
+OpenADSB Workflow
+
+```bash
+aerosim workflow openadsb \
+ --input-json inputs/your_adsb_data.json \
+ --lat 34.217411 \
+ --lon -118.491081 \
+ --radius 50 \
+ --interval 10
+```
+
+Steps:
+1. Convert raw ADS‑B JSON to CSV
+2. Filter based on geographic reference
+3. Convert to trajectory JSON
+4. Generate simulation scenario
+5. Create interactive visualization
+
+
+
+Artificial Track Workflow
+
+```bash
+aerosim workflow artificial \
+ --num-tracks 5 \
+ --maneuver elliptical \
+ --center-lat 34.217411 \
+ --center-lon -118.491081 \
+ --center-alt 1000 \
+ --separation 0.005 \
+ --time-delay 30 \
+ --num-points 20 \
+ --interval 10
+```
+
+Steps:
+1. Generate artificial tracks
+2. Save to CSV
+3. Convert to trajectory JSON
+4. Generate scenario
+5. Create visualization
+
+
+
+KML Workflow
+
+```bash
+aerosim workflow kml \
+ --input-kml inputs/my_track.kml \
+ --interval 10
+```
+
+Steps:
+1. Convert KML to CSV
+2. Convert to trajectory JSON
+3. Generate scenario
+4. Create visualization
+
+
+---
+
+
+