"""Analyze timestamped WGS84 trajectories with clustering and explicit rules.""" from __future__ import annotations import argparse import json import platform import sys import time from collections import defaultdict from datetime import UTC, datetime from importlib.metadata import PackageNotFoundError, version from pathlib import Path from typing import Any, Iterable import geopandas as gpd import matplotlib import numpy as np import pandas as pd from pyproj import CRS from shapely.geometry import LineString, Point, mapping from sklearn.cluster import DBSCAN from sklearn.preprocessing import StandardScaler matplotlib.use("Agg") import matplotlib.pyplot as plt from matplotlib.lines import Line2D from matplotlib.patches import Patch DEFAULT_THRESHOLDS = { "stop_speed_mps": 0.8, "stop_duration_seconds": 60.0, "route_deviation_m": 25.0, "route_deviation_duration_seconds": 20.0, "gathering_radius_m": 20.0, "gathering_duration_seconds": 30.0, "max_observation_gap_seconds": 60.0, } REQUIRED_COLUMNS = {"track_id", "entity_type", "timestamp", "longitude", "latitude"} EVENT_COLORS = { "stop": "#d55e00", "route_deviation": "#cc79a7", "restricted_zone": "#e69f00", "gathering": "#009e73", } EVENT_MARKERS = { "stop": "s", "route_deviation": "^", "restricted_zone": "D", "gathering": "P", } EVENT_SIZES = { "stop": 62, "route_deviation": 42, "restricted_zone": 92, "gathering": 68, } def parse_args() -> argparse.Namespace: root = Path(__file__).resolve().parents[2] parser = argparse.ArgumentParser(description="Analyze timestamped WGS84 trajectories.") parser.add_argument( "--input", type=Path, default=root / "shared" / "data" / "raw" / "15-trajectory-analysis", help="A .case.json manifest or a directory containing manifests.", ) parser.add_argument( "--output", type=Path, default=root / "shared" / "outputs" / "15-trajectory-analysis", ) parser.add_argument("--overwrite", action="store_true", help="Replace existing case outputs.") return parser.parse_args() def package_version(name: str) -> str: try: return version(name) except PackageNotFoundError: return "not-installed" def utc_text(value: pd.Timestamp | datetime) -> str: return value.to_pydatetime().astimezone(UTC).isoformat() if isinstance(value, pd.Timestamp) else value.astimezone(UTC).isoformat() def load_json(path: Path) -> dict[str, Any]: with path.open(encoding="utf-8") as stream: payload = json.load(stream) if not isinstance(payload, dict): raise ValueError(f"JSON root must be an object: {path}") return payload def resolve_input_path(manifest_path: Path, value: str) -> Path: path = Path(value) return path.resolve() if path.is_absolute() else (manifest_path.parent / path).resolve() def choose_metric_crs(longitude: float, latitude: float) -> CRS: zone = int((longitude + 180) // 6) + 1 epsg = (32600 if latitude >= 0 else 32700) + zone return CRS.from_epsg(epsg) def load_case(manifest_path: Path) -> tuple[dict[str, Any], pd.DataFrame, gpd.GeoDataFrame, gpd.GeoDataFrame, gpd.GeoDataFrame | None, int, CRS]: manifest = load_json(manifest_path) for field in ("case_id", "crs", "observations", "reference_routes", "zones"): if not manifest.get(field): raise ValueError(f"Manifest is missing '{field}': {manifest_path}") if manifest["crs"] != "EPSG:4326": raise ValueError("Demo v1 accepts only EPSG:4326 longitude/latitude observations.") observations_path = resolve_input_path(manifest_path, manifest["observations"]) routes_path = resolve_input_path(manifest_path, manifest["reference_routes"]) zones_path = resolve_input_path(manifest_path, manifest["zones"]) frame = pd.read_csv(observations_path) missing = REQUIRED_COLUMNS - set(frame.columns) if missing: raise ValueError(f"Observation CSV is missing columns: {', '.join(sorted(missing))}") raw_count = len(frame) if raw_count == 0: raise ValueError("Observation CSV is empty.") frame["track_id"] = frame["track_id"].astype(str).str.strip() frame["entity_type"] = frame["entity_type"].astype(str).str.strip() if (frame["track_id"] == "").any() or (frame["entity_type"] == "").any(): raise ValueError("track_id and entity_type cannot be empty.") frame["timestamp"] = pd.to_datetime(frame["timestamp"], utc=True, errors="raise") frame["longitude"] = pd.to_numeric(frame["longitude"], errors="raise") frame["latitude"] = pd.to_numeric(frame["latitude"], errors="raise") valid = frame["longitude"].between(-180, 180) & frame["latitude"].between(-90, 90) if not valid.all(): raise ValueError("longitude/latitude contains values outside WGS84 bounds.") frame = ( frame.sort_values(["track_id", "timestamp"]) .drop_duplicates(["track_id", "timestamp"], keep="last") .reset_index(drop=True) ) duplicate_count = raw_count - len(frame) if frame.groupby("track_id").size().min() < 2: raise ValueError("Every track must contain at least two distinct timestamps.") routes = gpd.read_file(routes_path) zones = gpd.read_file(zones_path) flyable_zones: gpd.GeoDataFrame | None = None if manifest.get("flyable_zones"): flyable_path = resolve_input_path(manifest_path, str(manifest["flyable_zones"])) flyable_zones = gpd.read_file(flyable_path) if flyable_zones.crs is None: raise ValueError("Flyable-zone GeoJSON must declare a CRS.") if "track_id" not in routes.columns: raise ValueError("Reference route GeoJSON must contain a track_id property.") if "zone_id" not in zones.columns or "zone_type" not in zones.columns: raise ValueError("Zone GeoJSON must contain zone_id and zone_type properties.") if routes.crs is None or zones.crs is None: raise ValueError("Route and zone GeoJSON files must declare a CRS.") expected = set(frame["track_id"]) missing_routes = expected - set(routes["track_id"].astype(str)) if missing_routes: raise ValueError(f"Missing reference routes for: {', '.join(sorted(missing_routes))}") metric_crs = choose_metric_crs(float(frame["longitude"].mean()), float(frame["latitude"].mean())) return ( manifest, frame, routes.to_crs(metric_crs), zones.to_crs(metric_crs), flyable_zones.to_crs(metric_crs) if flyable_zones is not None else None, duplicate_count, metric_crs, ) def true_runs(flags: list[bool], times: list[pd.Timestamp], max_gap: float) -> Iterable[tuple[int, int]]: start: int | None = None for index, flag in enumerate(flags): separated = index > 0 and (times[index] - times[index - 1]).total_seconds() > max_gap if flag and (start is None or separated): if start is not None: yield start, index - 1 start = index elif not flag and start is not None: yield start, index - 1 start = None if start is not None: yield start, len(flags) - 1 def make_event( case_id: str, event_type: str, track_ids: list[str], started: pd.Timestamp, ended: pd.Timestamp, point: Point, details: dict[str, Any], ) -> dict[str, Any]: duration = max(0.0, (ended - started).total_seconds()) return { "event_id": "", "case_id": case_id, "event_type": event_type, "track_ids": track_ids, "start_time": utc_text(started), "end_time": utc_text(ended), "duration_seconds": round(duration, 3), "longitude": round(point.x, 7), "latitude": round(point.y, 7), "details": details, } def analyze_tracks( case_id: str, frame: pd.DataFrame, routes_metric: gpd.GeoDataFrame, zones_metric: gpd.GeoDataFrame, metric_crs: CRS, thresholds: dict[str, float], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, gpd.GeoDataFrame]]: points = gpd.GeoDataFrame( frame.copy(), geometry=gpd.points_from_xy(frame["longitude"], frame["latitude"]), crs="EPSG:4326", ) points_metric = points.to_crs(metric_crs) route_map = {str(row.track_id): row.geometry for row in routes_metric.itertuples()} restricted = zones_metric[zones_metric["zone_type"].astype(str) == "restricted"] events: list[dict[str, Any]] = [] summaries: list[dict[str, Any]] = [] track_frames: dict[str, gpd.GeoDataFrame] = {} for track_id, group_indexes in points_metric.groupby("track_id", sort=True).groups.items(): track = points_metric.loc[group_indexes].sort_values("timestamp").copy().reset_index(drop=True) track_wgs84 = track.to_crs("EPSG:4326") track_frames[str(track_id)] = track_wgs84 times = list(track["timestamp"]) step_dt = track["timestamp"].diff().dt.total_seconds().fillna(0.0).to_numpy() step_distance = np.zeros(len(track), dtype=float) for index in range(1, len(track)): step_distance[index] = track.geometry.iloc[index - 1].distance(track.geometry.iloc[index]) step_speed = np.divide(step_distance, step_dt, out=np.zeros_like(step_distance), where=step_dt > 0) max_gap = thresholds["max_observation_gap_seconds"] slow_intervals = [ index > 0 and step_dt[index] <= max_gap and step_speed[index] <= thresholds["stop_speed_mps"] for index in range(len(track)) ] stop_seconds = 0.0 for start, end in true_runs(slow_intervals, times, max_gap): event_start = max(0, start - 1) duration = (times[end] - times[event_start]).total_seconds() if duration >= thresholds["stop_duration_seconds"]: stop_seconds += duration point = track_wgs84.geometry.iloc[event_start : end + 1].union_all().centroid events.append( make_event( case_id, "stop", [str(track_id)], times[event_start], times[end], point, {"max_speed_mps": round(float(step_speed[start : end + 1].max()), 3)}, ) ) route = route_map[str(track_id)] route_distances = np.array([geometry.distance(route) for geometry in track.geometry], dtype=float) deviated = list(route_distances > thresholds["route_deviation_m"]) deviation_seconds = 0.0 for start, end in true_runs(deviated, times, max_gap): duration = (times[end] - times[start]).total_seconds() if duration >= thresholds["route_deviation_duration_seconds"]: deviation_seconds += duration events.append( make_event( case_id, "route_deviation", [str(track_id)], times[start], times[end], track_wgs84.geometry.iloc[end], {"max_distance_m": round(float(route_distances[start : end + 1].max()), 3)}, ) ) restricted_seconds = 0.0 inside_by_zone: dict[str, list[bool]] = {} for zone in restricted.itertuples(): flags = [bool(zone.geometry.covers(geometry)) for geometry in track.geometry] inside_by_zone[str(zone.zone_id)] = flags for start, end in true_runs(flags, times, max_gap): duration = (times[end] - times[start]).total_seconds() restricted_seconds += duration events.append( make_event( case_id, "restricted_zone", [str(track_id)], times[start], times[end], track_wgs84.geometry.iloc[end], {"zone_id": str(zone.zone_id)}, ) ) duration_seconds = (times[-1] - times[0]).total_seconds() summaries.append( { "case_id": case_id, "track_id": str(track_id), "entity_type": str(track["entity_type"].iloc[0]), "point_count": len(track), "start_time": utc_text(times[0]), "end_time": utc_text(times[-1]), "duration_seconds": round(duration_seconds, 3), "distance_m": round(float(step_distance.sum()), 3), "average_speed_mps": round(float(step_distance.sum() / duration_seconds), 3), "max_speed_mps": round(float(step_speed.max()), 3), "route_deviation_ratio": round(float(np.mean(deviated)), 4), "restricted_zone_seconds": round(restricted_seconds, 3), "stop_seconds": round(stop_seconds, 3), "cluster_id": -1, "behavior_labels": "", } ) gathering_events = detect_gathering(case_id, points_metric, thresholds) events.extend(gathering_events) labels_by_track: dict[str, set[str]] = defaultdict(set) for event in events: for track_id in event["track_ids"]: labels_by_track[track_id].add(event["event_type"]) if len(summaries) >= 2: features = np.array( [ [ item["distance_m"], item["duration_seconds"], item["average_speed_mps"], item["route_deviation_ratio"], item["stop_seconds"] / max(item["duration_seconds"], 1), ] for item in summaries ] ) labels = DBSCAN(eps=1.35, min_samples=2).fit_predict(StandardScaler().fit_transform(features)) for item, label in zip(summaries, labels, strict=True): item["cluster_id"] = int(label) for item in summaries: item["behavior_labels"] = "|".join(sorted(labels_by_track[item["track_id"]])) or "normal" events.sort(key=lambda item: (item["start_time"], item["event_type"], item["track_ids"])) for index, event in enumerate(events, start=1): event["event_id"] = f"{case_id}-event-{index:03d}" return summaries, events, track_frames def detect_gathering( case_id: str, points_metric: gpd.GeoDataFrame, thresholds: dict[str, float] ) -> list[dict[str, Any]]: occurrences: dict[tuple[str, ...], list[tuple[pd.Timestamp, Point]]] = defaultdict(list) radius = thresholds["gathering_radius_m"] for timestamp, group in points_metric.groupby("timestamp"): rows = list(group.itertuples()) adjacency: dict[str, set[str]] = {str(row.track_id): set() for row in rows} geometries = {str(row.track_id): row.geometry for row in rows} for left_index, left in enumerate(rows): for right in rows[left_index + 1 :]: if left.geometry.distance(right.geometry) <= radius: adjacency[str(left.track_id)].add(str(right.track_id)) adjacency[str(right.track_id)].add(str(left.track_id)) remaining = set(adjacency) while remaining: seed = remaining.pop() component = {seed} queue = [seed] while queue: current = queue.pop() neighbors = adjacency[current] & remaining remaining -= neighbors component |= neighbors queue.extend(neighbors) if len(component) >= 2: members = tuple(sorted(component)) centroid = gpd.GeoSeries([geometries[item] for item in members], crs=points_metric.crs).union_all().centroid occurrences[members].append((timestamp, centroid)) events: list[dict[str, Any]] = [] max_gap = thresholds["max_observation_gap_seconds"] for members, values in occurrences.items(): values.sort(key=lambda item: item[0]) groups: list[list[tuple[pd.Timestamp, Point]]] = [[values[0]]] for value in values[1:]: if (value[0] - groups[-1][-1][0]).total_seconds() <= max_gap: groups[-1].append(value) else: groups.append([value]) for run in groups: duration = (run[-1][0] - run[0][0]).total_seconds() if duration < thresholds["gathering_duration_seconds"]: continue wgs84_point = gpd.GeoSeries([run[-1][1]], crs=points_metric.crs).to_crs("EPSG:4326").iloc[0] events.append( make_event( case_id, "gathering", list(members), run[0][0], run[-1][0], wgs84_point, {"member_count": len(members), "radius_m": radius}, ) ) return events def write_outputs( output_dir: Path, manifest: dict[str, Any], summaries: list[dict[str, Any]], events: list[dict[str, Any]], tracks: dict[str, gpd.GeoDataFrame], routes_wgs84: gpd.GeoDataFrame, zones_wgs84: gpd.GeoDataFrame, flyable_zones_wgs84: gpd.GeoDataFrame | None, metadata: dict[str, Any], ) -> None: output_dir.mkdir(parents=True, exist_ok=True) pd.DataFrame(summaries).to_csv(output_dir / "trajectory_summary.csv", index=False, encoding="utf-8-sig") (output_dir / "events.json").write_text( json.dumps({"case_id": manifest["case_id"], "events": events}, ensure_ascii=False, indent=2), encoding="utf-8", ) trajectory_features = [] summary_map = {item["track_id"]: item for item in summaries} for track_id, track in tracks.items(): properties = dict(summary_map[track_id]) trajectory_features.append( {"type": "Feature", "properties": properties, "geometry": mapping(LineString(track.geometry.tolist()))} ) write_geojson(output_dir / "trajectories.geojson", trajectory_features) event_features = [ { "type": "Feature", "properties": {key: value for key, value in event.items() if key not in {"longitude", "latitude"}}, "geometry": {"type": "Point", "coordinates": [event["longitude"], event["latitude"]]}, } for event in events ] write_geojson(output_dir / "events.geojson", event_features) route_features = json.loads(routes_wgs84.to_json())["features"] write_geojson(output_dir / "reference_routes.geojson", route_features) zone_features = json.loads(zones_wgs84.to_json())["features"] write_geojson(output_dir / "zones.geojson", zone_features) if flyable_zones_wgs84 is not None: flyable_features = json.loads(flyable_zones_wgs84.to_json())["features"] write_geojson(output_dir / "flyable_zones.geojson", flyable_features) render_map( output_dir / "analysis.png", tracks, routes_wgs84, zones_wgs84, flyable_zones_wgs84, events, manifest["case_id"], ) (output_dir / "run_metadata.json").write_text( json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8" ) def write_geojson(path: Path, features: list[dict[str, Any]]) -> None: path.write_text( json.dumps( {"type": "FeatureCollection", "name": path.stem, "crs": {"type": "name", "properties": {"name": "urn:ogc:def:crs:OGC:1.3:CRS84"}}, "features": features}, ensure_ascii=False, indent=2, ), encoding="utf-8", ) def render_map( path: Path, tracks: dict[str, gpd.GeoDataFrame], routes: gpd.GeoDataFrame, zones: gpd.GeoDataFrame, flyable_zones: gpd.GeoDataFrame | None, events: list[dict[str, Any]], case_id: str, ) -> None: figure, axis = plt.subplots(figsize=(10, 7), dpi=140) zones.plot(ax=axis, facecolor="#f0e442", edgecolor="#8c6d1f", alpha=0.28, linewidth=1.5) if flyable_zones is not None and not flyable_zones.empty: flyable_zones.plot(ax=axis, facecolor="#36a269", edgecolor="#167344", alpha=0.22, linewidth=1.3) palette = ["#0072b2", "#009e73", "#d55e00", "#cc79a7", "#56b4e9"] for index, (track_id, track) in enumerate(sorted(tracks.items())): color = palette[index % len(palette)] axis.plot(track.geometry.x, track.geometry.y, color=color, linewidth=2.2, marker="o", markersize=2.8, label=track_id) axis.annotate(track_id, (track.geometry.x.iloc[-1], track.geometry.y.iloc[-1]), fontsize=7, color=color) routes.plot(ax=axis, color="#444444", linestyle="--", linewidth=1.5, zorder=4, label="reference route") for event in events: event_type = event["event_type"] axis.scatter( [event["longitude"]], [event["latitude"]], marker=EVENT_MARKERS[event_type], s=EVENT_SIZES[event_type], color=EVENT_COLORS[event_type], edgecolor="white", linewidth=0.8, zorder=5, ) handles, labels = axis.get_legend_handles_labels() handles.append(Patch(facecolor="#f0e442", edgecolor="#8c6d1f", alpha=0.45, label="restricted area")) labels.append("restricted area") if flyable_zones is not None and not flyable_zones.empty: handles.append(Patch(facecolor="#36a269", edgecolor="#167344", alpha=0.45, label="flyable area")) labels.append("flyable area") present_event_types = {event["event_type"] for event in events} handles.extend( Line2D( [0], [0], marker=EVENT_MARKERS[name], color="none", markerfacecolor=color, markeredgecolor="white", markersize=8, label=name, ) for name, color in EVENT_COLORS.items() if name in present_event_types ) labels.extend(name for name in EVENT_COLORS if name in present_event_types) axis.legend(handles, labels, loc="best", fontsize=7, framealpha=0.9) axis.set_title(f"Trajectory analysis: {case_id}") axis.set_xlabel("Longitude (WGS84)") axis.set_ylabel("Latitude (WGS84)") axis.grid(alpha=0.18) axis.ticklabel_format(useOffset=False) visible_geometries = [geometry for track in tracks.values() for geometry in track.geometry] visible_geometries.extend(geometry for geometry in routes.geometry if geometry is not None) bounds = gpd.GeoSeries(visible_geometries, crs="EPSG:4326").total_bounds width = max(bounds[2] - bounds[0], 0.0005) height = max(bounds[3] - bounds[1], 0.0005) axis.set_xlim(bounds[0] - width * 0.08, bounds[2] + width * 0.08) axis.set_ylim(bounds[1] - height * 0.08, bounds[3] + height * 0.08) figure.tight_layout() figure.savefig(path) plt.close(figure) def process_manifest(manifest_path: Path, output_dir: Path) -> dict[str, Any]: started = time.perf_counter() manifest, frame, routes_metric, zones_metric, flyable_zones_metric, duplicates, metric_crs = load_case(manifest_path) thresholds = dict(DEFAULT_THRESHOLDS) for key, value in manifest.get("thresholds", {}).items(): if key not in thresholds: raise ValueError(f"Unsupported threshold: {key}") thresholds[key] = float(value) if thresholds[key] <= 0: raise ValueError(f"Threshold must be greater than zero: {key}") summaries, events, tracks = analyze_tracks( str(manifest["case_id"]), frame, routes_metric, zones_metric, metric_crs, thresholds ) elapsed = round(time.perf_counter() - started, 3) metadata = { "created_at": datetime.now(UTC).isoformat(), "case_id": manifest["case_id"], "input_manifest": str(manifest_path.resolve()), "input_count": int(len(frame)), "track_count": len(summaries), "dropped_duplicate_observations": duplicates, "event_count": len(events), "event_counts": {name: sum(event["event_type"] == name for event in events) for name in EVENT_COLORS}, "elapsed_seconds": elapsed, "device": "cpu", "python": platform.python_version(), "packages": { "geoai-py": package_version("geoai-py"), "pandas": package_version("pandas"), "geopandas": package_version("geopandas"), "shapely": package_version("shapely"), "pyproj": package_version("pyproj"), "scikit-learn": package_version("scikit-learn"), "matplotlib": package_version("matplotlib"), }, "model": { "behavior_recognition": "deterministic-threshold-rules-v1", "trajectory_grouping": "DBSCAN on standardized summary features", "pretrained_weights": None, }, "thresholds": thresholds, "output_crs": "EPSG:4326", "metric_crs": metric_crs.to_string(), "limitations": [ "This is rule-based behavior detection, not learned action recognition.", "Track identities must already be present; this demo does not associate detections across video frames.", "Gathering requires aligned timestamps and is sensitive to sampling gaps and GPS error.", "Thresholds are illustrative and require domain validation before operational use.", ], } write_outputs( output_dir, manifest, summaries, events, tracks, routes_metric.to_crs("EPSG:4326"), zones_metric.to_crs("EPSG:4326"), flyable_zones_metric.to_crs("EPSG:4326") if flyable_zones_metric is not None else None, metadata, ) return metadata def main() -> int: args = parse_args() input_path = args.input.resolve() output_root = args.output.resolve() if not input_path.exists(): print(f"Input does not exist: {input_path}", file=sys.stderr) return 2 manifests = [input_path] if input_path.is_file() else sorted(input_path.glob("*.case.json")) if not manifests: print(f"No .case.json manifests found: {input_path}", file=sys.stderr) return 2 try: cases = [(path, str(load_json(path).get("case_id", ""))) for path in manifests] if any(not case_id for _, case_id in cases): raise ValueError("Every manifest must define a non-empty case_id.") destinations = [output_root / case_id for _, case_id in cases] existing = [path for path in destinations if path.exists()] if existing and not args.overwrite: raise ValueError(f"Output already exists: {existing[0]}. Use --overwrite to replace it.") results = [process_manifest(path, destination) for (path, _), destination in zip(cases, destinations, strict=True)] aggregate = { "created_at": datetime.now(UTC).isoformat(), "input": str(input_path), "input_count": sum(item["input_count"] for item in results), "case_count": len(results), "track_count": sum(item["track_count"] for item in results), "dropped_duplicate_observations": sum( item["dropped_duplicate_observations"] for item in results ), "event_count": sum(item["event_count"] for item in results), "event_counts": { name: sum(item["event_counts"][name] for item in results) for name in EVENT_COLORS }, "elapsed_seconds": round(sum(item["elapsed_seconds"] for item in results), 3), "device": "cpu", "python": platform.python_version(), "packages": results[0]["packages"], "model": results[0]["model"], "thresholds_by_case": { item["case_id"]: item["thresholds"] for item in results }, "limitations": results[0]["limitations"], "cases": [ {"case_id": item["case_id"], "output": item["case_id"]} for item in results ], } output_root.mkdir(parents=True, exist_ok=True) (output_root / "run_metadata.json").write_text( json.dumps(aggregate, ensure_ascii=False, indent=2), encoding="utf-8" ) except (OSError, ValueError, json.JSONDecodeError, pd.errors.ParserError) as exc: print(f"Analysis failed: {exc}", file=sys.stderr) return 2 print( f"Processed {len(results)} case(s), {aggregate['track_count']} tracks and " f"{aggregate['event_count']} events in {aggregate['elapsed_seconds']}s" ) print(f"Outputs: {output_root}") return 0 if __name__ == "__main__": raise SystemExit(main())