from __future__ import annotations
|
|
import json
|
import subprocess
|
import sys
|
import tempfile
|
import unittest
|
from pathlib import Path
|
|
|
CAPABILITY_DIR = Path(__file__).resolve().parents[1]
|
GENERATOR = CAPABILITY_DIR / "generate_demo_inputs.py"
|
ANALYZER = CAPABILITY_DIR / "run_trajectory_analysis.py"
|
|
|
class TrajectoryDemoTests(unittest.TestCase):
|
def run_command(self, *arguments: str) -> subprocess.CompletedProcess[str]:
|
return subprocess.run(
|
[sys.executable, *arguments],
|
text=True,
|
capture_output=True,
|
encoding="utf-8",
|
check=False,
|
)
|
|
def test_normal_and_difficult_cases(self) -> None:
|
with tempfile.TemporaryDirectory() as temporary:
|
root = Path(temporary)
|
inputs = root / "inputs"
|
outputs = root / "outputs"
|
generated = self.run_command(str(GENERATOR), "--output", str(inputs))
|
self.assertEqual(generated.returncode, 0, generated.stderr)
|
analyzed = self.run_command(str(ANALYZER), "--input", str(inputs), "--output", str(outputs))
|
self.assertEqual(analyzed.returncode, 0, analyzed.stderr)
|
|
normal = json.loads((outputs / "normal" / "run_metadata.json").read_text(encoding="utf-8"))
|
difficult = json.loads((outputs / "difficult" / "run_metadata.json").read_text(encoding="utf-8"))
|
self.assertEqual(normal["track_count"], 2)
|
self.assertEqual(normal["event_count"], 0)
|
self.assertEqual(difficult["dropped_duplicate_observations"], 1)
|
for event_type in ("stop", "route_deviation", "restricted_zone", "gathering"):
|
self.assertGreaterEqual(difficult["event_counts"][event_type], 1)
|
for case_id in ("normal", "difficult"):
|
for filename in (
|
"trajectory_summary.csv",
|
"events.json",
|
"trajectories.geojson",
|
"events.geojson",
|
"reference_routes.geojson",
|
"zones.geojson",
|
"analysis.png",
|
"run_metadata.json",
|
):
|
self.assertGreater((outputs / case_id / filename).stat().st_size, 0)
|
|
def test_missing_required_column_fails_cleanly(self) -> None:
|
with tempfile.TemporaryDirectory() as temporary:
|
root = Path(temporary)
|
inputs = root / "inputs"
|
self.assertEqual(self.run_command(str(GENERATOR), "--output", str(inputs)).returncode, 0)
|
csv_path = inputs / "normal_observations.csv"
|
csv_path.write_text("track_id,timestamp,longitude,latitude\na,2026-01-01T00:00:00Z,1,1\n", encoding="utf-8")
|
result = self.run_command(
|
str(ANALYZER),
|
"--input",
|
str(inputs / "normal.case.json"),
|
"--output",
|
str(root / "outputs"),
|
)
|
self.assertEqual(result.returncode, 2)
|
self.assertIn("entity_type", result.stderr)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
