[uss_qualifier] Add flight intent transformations (#400)

* Add flight intent transformations

* Remove debugging message

monitoring/monitorlib/clients/flight_planning/flight_info_template.py

@@ -1,4 +1,4 @@
-from typing import Optional, Dict
+from typing import Optional, Dict, List
 
 from implicitdict import ImplicitDict
 
@@ -11,11 +11,13 @@
     BasicFlightPlanInformation,
     FlightInfo,
 )
+from monitoring.monitorlib.geo import LatLngPoint
 from monitoring.monitorlib.geotemporal import (
     Volume4DTemplateCollection,
     Volume4DCollection,
 )
 from monitoring.monitorlib.temporal import Time, TimeDuringTest
+from monitoring.monitorlib.transformations import Transformation
 from uas_standards.interuss.automated_testing.scd.v1 import api as scd_api
 
 
@@ -51,9 +53,16 @@ class FlightInfoTemplate(ImplicitDict):
     additional_information: Optional[dict]
     """Any information relevant to a particular jurisdiction or use case not described in the standard schema. The keys and values must be agreed upon between the test designers and USSs under test."""
 
+    transformations: Optional[List[Transformation]]
+    """If specified, transform this flight according to these transformations in order (after all templates are resolved)."""
+
     def resolve(self, times: Dict[TimeDuringTest, Time]) -> FlightInfo:
-        kwargs = {k: v for k, v in self.items()}
-        kwargs["basic_information"] = self.basic_information.resolve(times)
+        kwargs = {k: v for k, v in self.items() if k not in {"transformations"}}
+        basic_info = self.basic_information.resolve(times)
+        if "transformations" in self and self.transformations:
+            for xform in self.transformations:
+                basic_info.area = [v.transform(xform) for v in basic_info.area]
+        kwargs["basic_information"] = basic_info
         return ImplicitDict.parse(kwargs, FlightInfo)
 
     def to_scd_inject_request(

monitoring/monitorlib/geo.py

@@ -10,6 +10,12 @@
 from s2sphere import LatLng
 from scipy.interpolate import RectBivariateSpline as Spline
 import shapely.geometry
+
+from monitoring.monitorlib.transformations import (
+    Transformation,
+    RelativeTranslation,
+    AbsoluteTranslation,
+)
 from uas_standards.astm.f3548.v21 import api as f3548v21
 from uas_standards.astm.f3411.v19 import api as f3411v19
 from uas_standards.astm.f3411.v22a import api as f3411v22a
@@ -69,6 +75,10 @@ def from_f3411(
     def to_flight_planning_api(self) -> fp_api.LatLngPoint:
         return fp_api.LatLngPoint(lat=self.lat, lng=self.lng)
 
+    @staticmethod
+    def from_s2(p: s2sphere.LatLng) -> LatLngPoint:
+        return LatLngPoint(lat=p.lat().degrees, lng=p.lng().degrees)
+
     def as_s2sphere(self) -> s2sphere.LatLng:
         return s2sphere.LatLng.from_degrees(self.lat, self.lng)
 
@@ -89,7 +99,12 @@ def in_meters(self) -> float:
 
 
 class Polygon(ImplicitDict):
-    vertices: List[LatLngPoint]
+    vertices: Optional[List[LatLngPoint]]
+
+    def vertex_average(self) -> LatLngPoint:
+        lat = sum(p.lat for p in self.vertices) / len(self.vertices)
+        lng = sum(p.lng for p in self.vertices) / len(self.vertices)
+        return LatLngPoint(lat=lat, lng=lng)
 
     @staticmethod
     def from_coords(coords: List[Tuple[float, float]]) -> Polygon:
@@ -264,6 +279,81 @@ def intersects_vol3(self, vol3_2: Volume3D) -> bool:
 
         return footprint1.intersects(footprint2)
 
+    def transform(self, transformation: Transformation):
+        if (
+            "relative_translation" in transformation
+            and transformation.relative_translation
+        ):
+            return self.translate_relative(transformation.relative_translation)
+        elif (
+            "absolute_translation" in transformation
+            and transformation.absolute_translation
+        ):
+            return self.translate_absolute(transformation.absolute_translation)
+        raise ValueError(
+            f"No supported transformation defined (keys: {', '.join(transformation)})"
+        )
+
+    def translate_relative(self, translation: RelativeTranslation) -> Volume3D:
+        def offset(p0: LatLngPoint, p: LatLngPoint) -> LatLngPoint:
+            s2_p0 = p0.as_s2sphere()
+            xy = flatten(s2_p0, p.as_s2sphere())
+            if "meters_east" in translation and translation.meters_east:
+                xy = (xy[0] + translation.meters_east, xy[1])
+            if "meters_north" in translation and translation.meters_north:
+                xy = (xy[0], xy[1] + translation.meters_north)
+            p1 = LatLngPoint.from_s2(unflatten(s2_p0, xy))
+            if "degrees_east" in translation and translation.degrees_east:
+                p1.lng += translation.degrees_east
+            if "degrees_north" in translation and translation.degrees_north:
+                p1.lat += translation.degrees_north
+            return p1
+
+        kwargs = {k: v for k, v in self.items() if v is not None}
+        if self.outline_circle is not None:
+            kwargs["outline_circle"] = Circle(
+                center=offset(self.outline_circle.center, self.outline_circle.center),
+                radius=self.outline_circle.radius,
+            )
+        if self.outline_polygon is not None:
+            ref0 = self.outline_polygon.vertex_average()
+            vertices = [offset(ref0, p) for p in self.outline_polygon.vertices]
+            kwargs["outline_polygon"] = Polygon(vertices=vertices)
+        result = Volume3D(**kwargs)
+        if "meters_up" in translation and translation.meters_up:
+            if result.altitude_lower:
+                if result.altitude_lower.units == DistanceUnits.M:
+                    result.altitude_lower.value += translation.meters_up
+                else:
+                    raise NotImplementedError(
+                        f"Cannot yet translate meters_up with {result.altitude_lower.units} lower altitude units"
+                    )
+            if result.altitude_upper:
+                if result.altitude_upper.units == DistanceUnits.M:
+                    result.altitude_upper.value += translation.meters_up
+                else:
+                    raise NotImplementedError(
+                        f"Cannot yet translate meters_up with {result.altitude_lower.units} upper altitude units"
+                    )
+        return result
+
+    def translate_absolute(self, translation: AbsoluteTranslation) -> Volume3D:
+        new_center = LatLngPoint(
+            lat=translation.new_latitude, lng=translation.new_longitude
+        )
+        kwargs = {k: v for k, v in self.items() if v is not None}
+        if self.outline_circle is not None:
+            kwargs["outline_circle"] = Circle(
+                center=new_center, radius=self.outline_circle.radius
+            )
+        if self.outline_polygon is not None:
+            ref0 = self.outline_polygon.vertex_average().as_s2sphere()
+            xy = [flatten(ref0, p.as_s2sphere()) for p in self.outline_polygon.vertices]
+            ref1 = new_center.as_s2sphere()
+            vertices = [LatLngPoint.from_s2(unflatten(ref1, p)) for p in xy]
+            kwargs["outline_polygon"] = Polygon(vertices=vertices)
+        return Volume3D(**kwargs)
+
     @staticmethod
     def from_flight_planning_api(vol: fp_api.Volume3D) -> Volume3D:
         return ImplicitDict.parse(vol, Volume3D)

monitoring/monitorlib/geotemporal.py

@@ -4,9 +4,10 @@
 from datetime import datetime, timedelta
 from typing import Optional, List, Tuple, Dict
 
-import arrow
 from implicitdict import ImplicitDict, StringBasedTimeDelta
 import s2sphere as s2sphere
+
+from monitoring.monitorlib.transformations import Transformation
 from uas_standards.astm.f3548.v21 import api as f3548v21
 from uas_standards.interuss.automated_testing.flight_planning.v1 import api as fp_api
 from uas_standards.interuss.automated_testing.scd.v1 import api as interuss_scd_api
@@ -38,6 +39,9 @@ class Volume4DTemplate(ImplicitDict):
     altitude_upper: Optional[Altitude] = None
     """The maximum altitude at which the virtual user will fly while using this volume for their flight."""
 
+    transformations: Optional[List[Transformation]] = None
+    """If specified, transform this volume according to these transformations in order."""
+
     def resolve(self, times: Dict[TimeDuringTest, Time]) -> Volume4D:
         """Resolve Volume4DTemplate into concrete Volume4D."""
         # Make 3D volume
@@ -84,7 +88,13 @@ def resolve(self, times: Dict[TimeDuringTest, Time]) -> Volume4D:
         if time_end is not None:
             kwargs["time_end"] = time_end
 
-        return Volume4D(**kwargs)
+        result = Volume4D(**kwargs)
+
+        if self.transformations:
+            for xform in self.transformations:
+                result = result.transform(xform)
+
+        return result
 
 
 class Volume4D(ImplicitDict):
@@ -102,6 +112,11 @@ def offset_time(self, dt: timedelta) -> Volume4D:
             kwargs["time_end"] = self.time_end.offset(dt)
         return Volume4D(**kwargs)
 
+    def transform(self, transformation: Transformation) -> Volume4D:
+        kwargs = {k: v for k, v in self.items() if v is not None}
+        kwargs["volume"] = self.volume.transform(transformation)
+        return Volume4D(**kwargs)
+
     def intersects_vol4(self, vol4_2: Volume4D) -> bool:
         vol4_1 = self
         if vol4_1.time_end.datetime < vol4_2.time_start.datetime:

monitoring/monitorlib/transformations.py

@@ -0,0 +1,40 @@
+from typing import Optional
+
+from implicitdict import ImplicitDict
+
+
+class RelativeTranslation(ImplicitDict):
+    """Offset a geo feature by a particular amount."""
+
+    meters_east: Optional[float]
+    """Number of meters east to translate."""
+
+    meters_north: Optional[float]
+    """Number of meters north to translate."""
+
+    meters_up: Optional[float]
+    """Number of meters upward to translate."""
+
+    degrees_east: Optional[float]
+    """Number of degrees of longitude east to translate."""
+
+    degrees_north: Optional[float]
+    """Number of degrees of latitude north to translate."""
+
+
+class AbsoluteTranslation(ImplicitDict):
+    """Move a geo feature to a specified location."""
+
+    new_latitude: float
+    """The new latitude at which the feature should be located (degrees)."""
+
+    new_longitude: float
+    """The new longitude at which the feature should be located (degrees)."""
+
+
+class Transformation(ImplicitDict):
+    """A transformation to apply to a geotemporal feature.  Exactly one field must be specified."""
+
+    relative_translation: Optional[RelativeTranslation]
+
+    absolute_translation: Optional[AbsoluteTranslation]

monitoring/uss_qualifier/configurations/dev/f3548_self_contained.yaml

@@ -80,21 +80,42 @@ v1:
           resource_type: resources.flight_planning.FlightIntentsResource
           specification:
             file:
-              path: file://./test_data/che/flight_intents/conflicting_flights.yaml
+              path: file://./test_data/flight_intents/standard/conflicting_flights.yaml
+            transformations:
+              - relative_translation:
+                  # Put these flight intents in the DFW area
+                  degrees_north: 32.7181
+                  degrees_east: -96.7587
+
+                  # EGM96 geoid is 27.3 meters below the WGS84 ellipsoid at 32.7181, -96.7587
+                  # Ground level starts at roughly 120m above the EGM96 geoid
+                  # Therefore, ground level is at roughly 93m above the WGS84 ellipsoid
+                  meters_up: 93
 
         # Details of flights with invalid operational intents (used in flight intent validation scenario)
         invalid_flight_intents:
           resource_type: resources.flight_planning.FlightIntentsResource
           specification:
             intent_collection:
-              $ref: test_data.che.flight_intents.invalid_flight_intents
+              $ref: test_data.flight_intents.standard.invalid_flight_intents
+            transformations:
+              - relative_translation:
+                  degrees_north: 32.7181
+                  degrees_east: -96.7587
+                  meters_up: 93
 
         # Details of non-conflicting flights (used in data validation scenario)
         non_conflicting_flights:
           resource_type: resources.flight_planning.FlightIntentsResource
           specification:
             intent_collection:
-              $ref: file://../../test_data/usa/kentland/flight_intents/non_conflicting.yaml
+              # Note that $refs are relative to the file with the $ref (this one, in this case)
+              $ref: file://../../test_data/flight_intents/standard/non_conflicting.yaml
+            transformations:
+              - relative_translation:
+                  degrees_north: 32.7181
+                  degrees_east: -96.7587
+                  meters_up: 93
 
         # Location of DSS instance that can be used to verify flight planning outcomes
         dss:

monitoring/uss_qualifier/configurations/dev/library/resources.yaml

@@ -132,17 +132,27 @@ che_conflicting_flights:
   resource_type: resources.flight_planning.FlightIntentsResource
   specification:
     file:
-      path: file://./test_data/che/flight_intents/conflicting_flights.yaml
+      path: file://./test_data/flight_intents/standard/conflicting_flights.yaml
       # Note that this hash_sha512 field can be safely deleted if the content changes
-      hash_sha512: 26ee66a5065e555512f8b1e354334678dfe1614c6fbba4898a1541e6306341620e96de8b48e4095c7b03ab6fd58d0aeeee9e69cf367e1b7346e0c5f287460792
+      hash_sha512: b5432d496928aaa1876acc754e9ffa12f407809a014fa90e23f450c013fb20e2321328d48a419bc129276f7e9e26002c0fea6fec9baf3952b60daec6197de6b7
+    transformations:
+      - relative_translation:
+          degrees_north: 46.9748
+          degrees_east: 7.4774
+          meters_up: 605
 
 che_invalid_flight_intents:
   $content_schema: monitoring/uss_qualifier/resources/definitions/ResourceDeclaration.json
   resource_type: resources.flight_planning.FlightIntentsResource
   specification:
     intent_collection:
       # Note that $refs may use package-based paths
-      $ref: test_data.che.flight_intents.invalid_flight_intents
+      $ref: test_data.flight_intents.standard.invalid_flight_intents
+    transformations:
+      - relative_translation:
+          degrees_north: 46.9748
+          degrees_east: 7.4774
+          meters_up: 605
 
 che_general_flight_auth_flights:
   $content_schema: monitoring/uss_qualifier/resources/definitions/ResourceDeclaration.json
@@ -158,7 +168,12 @@ che_non_conflicting_flights:
   specification:
     file:
       # Note that ExternalFile paths may be package-based
-      path: test_data.che.flight_intents.non_conflicting
+      path: test_data.flight_intents.standard.non_conflicting
+    transformations:
+      - relative_translation:
+          degrees_north: 46.9748
+          degrees_east: 7.4774
+          meters_up: 605
 
 # ===== General flight authorization =====
 

monitoring/uss_qualifier/resources/flight_planning/flight_intent.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from typing import Optional, Dict
+from typing import Optional, Dict, List
 
 import arrow
 
@@ -10,6 +10,7 @@
     FlightInfoTemplate,
 )
 from monitoring.monitorlib.temporal import Time, TimeDuringTest
+from monitoring.monitorlib.transformations import Transformation
 
 from monitoring.uss_qualifier.resources.files import ExternalFile
 from monitoring.uss_qualifier.resources.overrides import apply_overrides
@@ -71,6 +72,9 @@ class FlightIntentCollection(ImplicitDict):
     intents: Dict[FlightIntentID, FlightIntentCollectionElement]
     """Flight planning actions that users want to perform."""
 
+    transformations: Optional[List[Transformation]]
+    """Transformations to append to all FlightInfoTemplates."""
+
     def resolve(self) -> Dict[FlightIntentID, FlightInfoTemplate]:
         """Resolve the underlying delta flight intents."""
 
@@ -114,6 +118,16 @@ def resolve(self) -> Dict[FlightIntentID, FlightInfoTemplate]:
                     + ", ".join(i_id for i_id in unprocessed_intent_ids)
                 )
 
+        if "transformations" in self and self.transformations:
+            for v in processed_intents.values():
+                xforms = (
+                    v.transformations.copy()
+                    if v.has_field_with_value("transformations")
+                    else []
+                )
+                xforms.extend(self.transformations)
+                v.transformations = xforms
+
         return processed_intents
 
 
@@ -125,3 +139,6 @@ class FlightIntentsSpecification(ImplicitDict):
 
     file: Optional[ExternalFile]
     """Location of file to load, containing a FlightIntentCollection"""
+
+    transformations: Optional[List[Transformation]]
+    """Transformations to apply to all flight intents' 4D volumes after resolution (if specified)"""