[uss_qualifier/resource] Add base class for geospatial modifier

.basedpyright/baseline.json

@@ -19113,14 +19113,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportAttributeAccessIssue",
-                "range": {
-                    "startColumn": 44,
-                    "endColumn": 58,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportCallIssue",
                 "range": {

monitoring/uss_qualifier/resources/dev/__init__.py

@@ -4,3 +4,5 @@
     NumberGeneratorModifierResource as NumberGeneratorModifierResource,
 )
 from .test_modifier import NumberGeneratorResource as NumberGeneratorResource
+from .test_modifier import TestSquareModifier as TestSquareModifier
+from .test_modifier import TestSquareResource as TestSquareResource

monitoring/uss_qualifier/resources/dev/test_modifier.py

@@ -1,5 +1,10 @@
 from implicitdict import ImplicitDict
 
+from monitoring.monitorlib.geo import LatLngBoundingBox
+from monitoring.uss_qualifier.resources.geospatial import (
+    GeospatialResource,
+    TriangularCascadeSoutheastResource,
+)
 from monitoring.uss_qualifier.resources.resource import (
     Resource,
     ResourceProvidingResource,
@@ -70,3 +75,57 @@ def provide_resource_for(self, **kwargs) -> NumberGeneratorResource:
             ),
             resource_origin=self._modified_resource_origin(index),
         )
+
+
+class TestSquareSpecification(ImplicitDict):
+    lat_center: float
+    lng_center: float
+
+
+class TestSquareResource(GeospatialResource, Resource[TestSquareSpecification]):
+    """1km x 1km square centered at (lat_center, lng_center). Used for unit tests."""
+
+    SQUARE_SIDE_M = 1000.0
+
+    _spec: TestSquareSpecification
+
+    def __init__(
+        self,
+        specification: TestSquareSpecification,
+        resource_origin: str,
+    ):
+        super().__init__(specification, resource_origin)
+        self._spec = specification
+
+    def get_extents(self) -> LatLngBoundingBox:
+        point = LatLngBoundingBox(
+            lat_min=self._spec.lat_center,
+            lat_max=self._spec.lat_center,
+            lng_min=self._spec.lng_center,
+            lng_max=self._spec.lng_center,
+        )
+        return point.expand(
+            north_meters=self.SQUARE_SIDE_M / 2,
+            east_meters=self.SQUARE_SIDE_M / 2,
+            south_meters=self.SQUARE_SIDE_M / 2,
+            west_meters=self.SQUARE_SIDE_M / 2,
+        )
+
+    def move(self, meters_east: float, meters_north: float) -> "TestSquareResource":
+        shifted = self.get_extents().expand(
+            north_meters=meters_north,
+            east_meters=meters_east,
+            south_meters=-meters_north,
+            west_meters=-meters_east,
+        )
+        return TestSquareResource(
+            TestSquareSpecification(
+                lat_center=(shifted.lat_min + shifted.lat_max) / 2,
+                lng_center=(shifted.lng_min + shifted.lng_max) / 2,
+            ),
+            resource_origin=self.resource_origin,
+        )
+
+
+class TestSquareModifier(TriangularCascadeSoutheastResource[TestSquareResource]):
+    pass

monitoring/uss_qualifier/resources/geospatial.py

@@ -0,0 +1,75 @@
+from abc import ABC, abstractmethod
+from math import isqrt
+from typing import Self
+
+from implicitdict import ImplicitDict
+
+from monitoring.monitorlib.geo import LatLngBoundingBox, flatten
+from monitoring.uss_qualifier.resources.resource import (
+    Resource,
+    ResourceProvidingResource,
+)
+
+
+class GeospatialResource(Resource, ABC):
+    @abstractmethod
+    def get_extents(self) -> LatLngBoundingBox:
+        pass
+
+    @abstractmethod
+    def move(self, meters_east: float, meters_north: float) -> Self:
+        """Return a copy of this resource that has been moved the specified number of meters east and north."""
+        pass
+
+
+class TriangularCascadeSoutheastSpecification(ImplicitDict):
+    meters_east_margin: float
+    """Modify the resource by moving it this far along the east-west axis to separate it from other modification instances."""
+
+    meters_north_margin: float
+    """Modify the resource by moving it this far along the north-south axis to separate it from other modification instances."""
+
+
+class TriangularCascadeSoutheastResource[GeospatialResourceType: GeospatialResource](
+    ResourceProvidingResource[
+        TriangularCascadeSoutheastSpecification, GeospatialResourceType
+    ]
+):
+    """Provides modified copies of a base geospatial resource which are offset east and south of the original resource."""
+
+    base_resource: GeospatialResourceType
+
+    def __init__(
+        self,
+        specification: TriangularCascadeSoutheastSpecification,
+        resource_origin: str,
+        base_resource: GeospatialResourceType,
+    ):
+        super().__init__(specification, resource_origin)
+        self._spec = specification
+        self.base_resource = base_resource
+
+    def provide_resource_for(self, **kwargs) -> GeospatialResourceType:
+
+        if "index" not in kwargs:
+            raise ValueError("Need an index")
+
+        index = kwargs["index"]
+        assert isinstance(index, int)
+
+        # Make a grid based on index:
+        #      x ->
+        # y  0 1 3 6
+        # |  2 4 7
+        # v  5 8
+        #    9
+        k = (isqrt(1 + 8 * index) - 1) // 2
+        offset = index - k * (k + 1) // 2
+        x = k - offset
+        y = offset
+
+        rect = self.base_resource.get_extents().to_latlngrect()
+        width_m, height_m = flatten(rect.lo(), rect.hi())
+        width_m += self._spec.meters_east_margin
+        height_m += self._spec.meters_north_margin
+        return self.base_resource.move(x * width_m, y * height_m)

monitoring/uss_qualifier/resources/geospatial_test.py

@@ -0,0 +1,63 @@
+import unittest
+
+from monitoring.monitorlib.geo import area_of_latlngrect
+from monitoring.uss_qualifier.resources.definitions import (
+    ResourceDeclaration,
+    ResourceID,
+)
+from monitoring.uss_qualifier.resources.dev.test_modifier import (
+    TestSquareSpecification,
+)
+from monitoring.uss_qualifier.resources.geospatial import (
+    TriangularCascadeSoutheastSpecification,
+)
+from monitoring.uss_qualifier.resources.resource import create_resources
+
+
+class TestGeospatialModifier(unittest.TestCase):
+    def _build_declarations(self) -> dict[ResourceID, ResourceDeclaration]:
+        return {
+            "square": ResourceDeclaration(
+                resource_type="resources.dev.TestSquareResource",
+                specification=TestSquareSpecification(lat_center=46.5, lng_center=6.5),
+            ),
+            "square_modifier": ResourceDeclaration(
+                resource_type="resources.dev.TestSquareModifier",
+                specification=TriangularCascadeSoutheastSpecification(
+                    meters_east_margin=1000, meters_north_margin=1000
+                ),
+                dependencies={
+                    "base_resource": "square",
+                },
+            ),
+        }
+
+    def test_overlap_only_for_same_index(self):
+        resources = create_resources(self._build_declarations(), "test", True)
+        modifier = resources["square_modifier"]
+
+        extents = [
+            modifier.provide_resource_for(index=i).get_extents() for i in range(11)
+        ]
+        square_area = (
+            resources["square"].SQUARE_SIDE_M * resources["square"].SQUARE_SIDE_M
+        )
+
+        for i in range(11):
+            for j in range(11):
+                rect_i = extents[i].to_latlngrect()
+                rect_j = extents[j].to_latlngrect()
+                overlap = area_of_latlngrect(rect_i.intersection(rect_j))
+                if i == j:
+                    assert (
+                        overlap > 0.99 * square_area
+                    ), (  # Use 99% to compensate for errors
+                        f"index {i}: self-overlap area {overlap:.2f}m² "
+                        f"expected ~{square_area:.2f}m²"
+                    )
+                else:
+                    assert (
+                        overlap < 0.01 * square_area
+                    ), (  # Use 1% to compensate for errors
+                        f"indices {i},{j}: unexpected overlap area {overlap:.2f}m²"
+                    )

monitoring/uss_qualifier/resources/resource.py

@@ -179,15 +179,21 @@ def get_resource_types(
 
     constructor_signature = get_type_hints(resource_type.__init__)
 
-    # Resolve generic type vars
+    # Resolve generic type vars, walking up the inheritance chain
+    def _collect(cls: type, mapping: dict) -> None:
+        for base in getattr(cls, "__orig_bases__", ()):
+            origin = get_origin(base)
+            if origin is None:
+                continue
+            params = getattr(origin, "__parameters__", ())
+            for param, arg in zip(params, get_args(base)):
+                resolved = mapping.get(arg, arg)
+                if not isinstance(resolved, TypeVar):
+                    mapping[param] = resolved
+            _collect(origin, mapping)
+
     typevar_map: dict = {}
-    for base in getattr(resource_type, "__orig_bases__", ()):
-        params = getattr(get_origin(base), "__type_params__", None) or getattr(
-            get_origin(base), "__parameters__", ()
-        )
-        for param, arg in zip(params, get_args(base)):
-            if not isinstance(arg, TypeVar):
-                typevar_map[param] = arg
+    _collect(resource_type, typevar_map)
     constructor_signature = {
         name: typevar_map.get(t, t) for name, t in constructor_signature.items()
     }

schemas/manage_type_schemas.py

@@ -87,6 +87,29 @@ def _make_type_schemas(
                     )
 
 
+def _resolve_resource_spec_type(cls: type) -> type:
+    """Find the spec type bound to Resource[Spec] for a Resource subclass,
+    resolving TypeVars through intermediate generic bases (e.g., ResourceModifier)."""
+
+    def walk(c: type, subst: dict):
+        for base in getattr(c, "__orig_bases__", ()):
+            origin = get_origin(base)
+            if origin is None:
+                continue
+            args = tuple(subst.get(a, a) for a in get_args(base))
+            if origin is Resource:
+                return args[0]
+            result = walk(origin, dict(zip(origin.__parameters__, args)))
+            if result is not None:
+                return result
+        return None
+
+    result = walk(cls, {})
+    if result is None:
+        raise ValueError(f"Could not resolve Resource specification type for {cls}")
+    return result
+
+
 def _find_specifications(
     module,
     repo: dict[str, type[ImplicitDict]],
@@ -107,7 +130,7 @@ def _find_specifications(
             if issubclass(member, Resource) and member != Resource:
                 if inspect.isabstract(member):
                     continue
-                spec_type = get_args(member.__orig_bases__[0])[0]
+                spec_type = _resolve_resource_spec_type(member)
                 repo[fullname(spec_type)] = spec_type
             elif issubclass(member, ActionGenerator) and member != ActionGenerator:
                 spec_type = get_args(member.__orig_bases__[0])[0]

schemas/monitoring/uss_qualifier/resources/dev/test_modifier/TestSquareSpecification.json

@@ -0,0 +1,22 @@
+{
+  "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/uss_qualifier/resources/dev/test_modifier/TestSquareSpecification.json",
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "description": "monitoring.uss_qualifier.resources.dev.test_modifier.TestSquareSpecification, as defined in monitoring/uss_qualifier/resources/dev/test_modifier.py",
+  "properties": {
+    "$ref": {
+      "description": "Path to content that replaces the $ref",
+      "type": "string"
+    },
+    "lat_center": {
+      "type": "number"
+    },
+    "lng_center": {
+      "type": "number"
+    }
+  },
+  "required": [
+    "lat_center",
+    "lng_center"
+  ],
+  "type": "object"
+}

schemas/monitoring/uss_qualifier/resources/geospatial/TriangularCascadeSoutheastSpecification.json

@@ -0,0 +1,24 @@
+{
+  "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/uss_qualifier/resources/geospatial/TriangularCascadeSoutheastSpecification.json",
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "description": "monitoring.uss_qualifier.resources.geospatial.TriangularCascadeSoutheastSpecification, as defined in monitoring/uss_qualifier/resources/geospatial.py",
+  "properties": {
+    "$ref": {
+      "description": "Path to content that replaces the $ref",
+      "type": "string"
+    },
+    "meters_east_margin": {
+      "description": "Modify the resource by moving it this far along the east-west axis to separate it from other modification instances.",
+      "type": "number"
+    },
+    "meters_north_margin": {
+      "description": "Modify the resource by moving it this far along the north-south axis to separate it from other modification instances.",
+      "type": "number"
+    }
+  },
+  "required": [
+    "meters_east_margin",
+    "meters_north_margin"
+  ],
+  "type": "object"
+}