from __future__ import annotations
|
|
import json
|
import gzip
|
import subprocess
|
import sys
|
import tempfile
|
import unittest
|
from pathlib import Path
|
from zipfile import ZipFile
|
|
from openpyxl import Workbook
|
|
|
CAPABILITY_DIR = Path(__file__).resolve().parents[1]
|
PREPARER = CAPABILITY_DIR / "prepare_real_flight.py"
|
ANALYZER = CAPABILITY_DIR / "run_trajectory_analysis.py"
|
|
|
class PrepareRealFlightTests(unittest.TestCase):
|
def test_prepares_a_runnable_single_flight_case(self) -> None:
|
with tempfile.TemporaryDirectory() as temporary:
|
root = Path(temporary)
|
raw = root / "raw"
|
(raw / "tracks").mkdir(parents=True)
|
(raw / "routes").mkdir()
|
(raw / "areas").mkdir()
|
workbook = Workbook()
|
sheet = workbook.active
|
sheet.append([
|
"\u98de\u884c\u7c7b\u578b", "\u98de\u884c\u4efb\u52a1ID", "\u7eac\u5ea6", "\u7ecf\u5ea6", "\u7edd\u5bf9\u9ad8\u5ea6(m)", "\u5b9e\u65f6\u771f\u9ad8(m)", "\u521b\u5efa\u65f6\u95f4"
|
])
|
sheet.append(["\u822a\u7ebf\u98de\u884c", 42, 28.0, 118.0, 100.0, 30.0, "2026-01-01 08:00:00"])
|
sheet.append(["\u822a\u7ebf\u98de\u884c", 42, 28.0001, 118.0001, 100.0, 30.0, "2026-01-01 08:00:10"])
|
workbook.save(raw / "tracks" / "flight.xlsx")
|
kmz = raw / "routes" / "route.kmz"
|
wpml = """<?xml version='1.0' encoding='UTF-8'?><kml xmlns='http://www.opengis.net/kml/2.2' xmlns:wpml='http://www.dji.com/wpmz/1.0.5'><Document><Folder><Placemark><Point><coordinates>118.0,28.0</coordinates></Point><wpml:index>0</wpml:index></Placemark><Placemark><Point><coordinates>118.0001,28.0001</coordinates></Point><wpml:index>1</wpml:index></Placemark></Folder></Document></kml>"""
|
with ZipFile(kmz, "w") as archive:
|
archive.writestr("wpmz/waylines.wpml", wpml)
|
zones = {
|
"type": "FeatureCollection",
|
"crs": {"type": "name", "properties": {"name": "urn:ogc:def:crs:OGC:1.3:CRS84"}},
|
"features": [{"type": "Feature", "properties": {"area_id": 1, "name": "test", "city": "test", "level": 2, "height": 120, "radius": 1000}, "geometry": {"type": "Point", "coordinates": [118.0, 28.0]}}],
|
}
|
(raw / "areas" / "zones.geojson").write_text(json.dumps(zones), encoding="utf-8")
|
ring = [(118.0, 28.0), (118.001, 28.0), (118.001, 28.001), (118.0, 28.001), (118.0, 28.0)]
|
encoded = [value for point in ring for value in (round(point[0] * 1e7), round(point[1] * 1e7))] + [1]
|
(raw / "areas" / "flyable.gzip").write_bytes(gzip.compress(__import__("array").array("i", encoded).tobytes()))
|
prepared = root / "prepared"
|
result = subprocess.run(
|
[sys.executable, str(PREPARER), "--raw-dir", str(raw), "--output", str(prepared), "--case-id", "sample"],
|
capture_output=True,
|
encoding="utf-8",
|
text=True,
|
check=False,
|
)
|
self.assertEqual(result.returncode, 0, result.stderr)
|
observations = (prepared / "observations.csv").read_text(encoding="utf-8-sig")
|
self.assertIn("flight_phase", observations)
|
self.assertIn("2026-01-01T00:00:00Z", observations)
|
zone_payload = json.loads((prepared / "zones.geojson").read_text(encoding="utf-8"))
|
self.assertEqual(zone_payload["features"][0]["properties"]["zone_type"], "restricted")
|
flyable_payload = json.loads((prepared / "flyable_zones.geojson").read_text(encoding="utf-8"))
|
self.assertEqual(flyable_payload["features"][0]["properties"]["zone_type"], "flyable")
|
analyzed = subprocess.run(
|
[sys.executable, str(ANALYZER), "--input", str(prepared / "sample.case.json"), "--output", str(root / "outputs")],
|
capture_output=True,
|
encoding="utf-8",
|
text=True,
|
check=False,
|
)
|
self.assertEqual(analyzed.returncode, 0, analyzed.stderr)
|
self.assertTrue((root / "outputs" / "sample" / "zones.geojson").is_file())
|
self.assertTrue((root / "outputs" / "sample" / "flyable_zones.geojson").is_file())
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
