[uss_qualifier] Generate FeatureCheckTable at runtime when configured

monitoring/uss_qualifier/configurations/dev/geospatial_comprehension.yaml

@@ -4,15 +4,34 @@ v1:
     resources:
       resource_declarations:
         example_feature_check_table: {$ref: 'library/resources.yaml#/example_feature_check_table'}
+        generated_feature_check_table: {$ref: 'library/resources.yaml#/generated_feature_check_table'}
 
         utm_auth: {$ref: 'library/environment.yaml#/utm_auth'}
         geospatial_info_provider: {$ref: 'library/environment.yaml#/geospatial_info_provider'}
     action:
-      test_scenario:
-        scenario_type: scenarios.interuss.geospatial_map.GeospatialFeatureComprehension
+      test_suite:
         resources:
           geospatial_info_provider: geospatial_info_provider
-          table: example_feature_check_table
+          example_feature_check_table: example_feature_check_table
+          generated_feature_check_table: generated_feature_check_table
+        suite_definition:
+          name: Geospatial comprehension CI suite
+          resources:
+            geospatial_info_provider: resources.geospatial_info.GeospatialInfoProviderResource
+            example_feature_check_table: resources.interuss.geospatial_map.FeatureCheckTableResource
+            generated_feature_check_table: resources.interuss.geospatial_map.FeatureCheckTableResource
+          actions:
+          - test_scenario:
+              scenario_type: scenarios.interuss.geospatial_map.GeospatialFeatureComprehension
+              resources:
+                geospatial_info_provider: geospatial_info_provider
+                table: example_feature_check_table
+          - test_scenario:
+              scenario_type: scenarios.interuss.geospatial_map.GeospatialFeatureComprehension
+              resources:
+                geospatial_info_provider: geospatial_info_provider
+                table: generated_feature_check_table
+
     execution:
       stop_fast: true
   artifacts:

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

@@ -522,6 +522,17 @@ example_feature_check_table:
               duration: 5m
           expected_result: Advise
 
+generated_feature_check_table:
+  $content_schema: monitoring/uss_qualifier/resources/definitions/ResourceDeclaration.json
+  resource_type: resources.interuss.geospatial_map.FeatureCheckTableResource
+  specification:
+    generate_at_runtime:
+      dict_resources:
+        geojson_example:
+          path: test_data.aus.incidents
+      jsonnet_script:
+        path: test_data.aus.incident_translator
+
 # ===== mock_uss behavioral control =====
 
 locality_che:

monitoring/uss_qualifier/resources/interuss/geospatial_map/README.md

@@ -0,0 +1,12 @@
+# geospatial_map automated testing
+
+The resources in this folder are primarily intended to be used in the [scenarios.interuss.geospatial_map.GeospatialFeatureComprehension](../../../scenarios/interuss/geospatial_map/geospatial_feature_comprehension.md) test scenario.
+
+When designing a FeatureCheckTable that is generated at runtime, the [generate_featurechecktable.py](./generate_featurechecktable.py) utility can be used to ease development.  It accepts a path to a dict-like file containing a [FeatureCheckTableSpecification](./definitions.py) -- subelements of a file can be selected with an anchor like `#/foo/bar` -- and prints the generated FeatureCheckTable in YAML.  So, for instance, to see the FeatureCheckTable generated by the `generated_feature_check_table` resource in the [library resources.yaml](../../../configurations/dev/library/resources.yaml), run (from the repo root):
+
+```shell
+make image
+docker container run -it  interuss/monitoring uv run \
+  ./uss_qualifier/resources/interuss/geospatial_map/generate_featurechecktable.py \
+  --config-path file://./uss_qualifier/configurations/dev/library/resources.yaml#/generated_feature_check_table/specification 
+```

monitoring/uss_qualifier/resources/interuss/geospatial_map/feature_check_table.py

@@ -1,13 +1,85 @@
+import json
+import os
+
+import _jsonnet
+import arrow
 from implicitdict import ImplicitDict
 
+from monitoring.uss_qualifier.resources.files import (
+    ExternalFile,
+    load_content,
+    load_dict,
+)
 from monitoring.uss_qualifier.resources.interuss.geospatial_map.definitions import (
     FeatureCheckTable,
 )
 from monitoring.uss_qualifier.resources.resource import Resource
 
 
+class FeatureCheckTableGenerationSpecification(ImplicitDict):
+    dict_resources: dict[str, ExternalFile] | None
+    """External dict-like content (.json, .yaml, .jsonnet) to load and make available to the script.
+    Key defines the name of the resource accessible to the script.
+    """
+
+    jsonnet_script: ExternalFile
+    """Source of Jsonnet "script" that produces a FeatureCheckTable.
+
+    This converter may access resources with std.extVars("resource_name").  Also, std.extVars("now") returns the current
+    datetime as an ISO string.  The return value of the Jsonnet must be an object following the FeatureCheckTable
+    schema.  `import` is not allowed.  The following additional functions are available:
+      * std.native("timestamp_of")(s: str) -> float
+        * s: Datetime string
+        * Returns: Seconds past epoch
+    """
+
+
 class FeatureCheckTableSpecification(ImplicitDict):
-    table: FeatureCheckTable
+    table: FeatureCheckTable | None
+    """Statically-defined feature check table"""
+
+    generate_at_runtime: FeatureCheckTableGenerationSpecification | None
+    """Generate feature check table at runtime"""
+
+
+def generate_feature_check_table(
+    spec: FeatureCheckTableGenerationSpecification,
+) -> FeatureCheckTable:
+    # Load dict resources
+    if "dict_resources" in spec and spec.dict_resources:
+        resources = {
+            k: json.dumps(load_dict(v)) for k, v in spec.dict_resources.items()
+        }
+    else:
+        resources = {}
+
+    # Useful functions not included in Jsonnet
+    native_callbacks = {"timestamp_of": (("s",), lambda s: arrow.get(s).timestamp())}
+
+    # Useful variables not included in Jsonnet
+    ext_vars = {"now": arrow.utcnow().isoformat()}
+
+    # Behavior when Jsonnet attempts to import something
+    def jsonnet_import_callback(folder: str, rel: str) -> tuple[str, bytes]:
+        raise ValueError("Jsonnet to generate FeatureCheckTable may not use `import`")
+
+    # Load Jsonnet "script"
+    jsonnet_string = load_content(spec.jsonnet_script)
+
+    # Run Jsonnet "script"
+    file_name = os.path.split(spec.jsonnet_script.path)[-1]
+    json_str = _jsonnet.evaluate_snippet(
+        file_name,
+        jsonnet_string,
+        ext_codes=resources,
+        ext_vars=ext_vars,
+        import_callback=jsonnet_import_callback,
+        native_callbacks=native_callbacks,  # pyright: ignore [reportArgumentType]
+    )
+
+    # Parse output into FeatureCheckTable
+    dict_content = json.loads(json_str)
+    return ImplicitDict.parse(dict_content, FeatureCheckTable)
 
 
 class FeatureCheckTableResource(Resource[FeatureCheckTableSpecification]):
@@ -17,4 +89,11 @@ def __init__(
         self, specification: FeatureCheckTableSpecification, resource_origin: str
     ):
         super().__init__(specification, resource_origin)
-        self.table = specification.table
+        if "table" in specification and specification.table:
+            self.table = specification.table
+        elif (
+            "generate_at_runtime" in specification and specification.generate_at_runtime
+        ):
+            self.table = generate_feature_check_table(specification.generate_at_runtime)
+        else:
+            raise ValueError("No means to define FeatureCheckTable was specified")

monitoring/uss_qualifier/resources/interuss/geospatial_map/generate_featurechecktable.py

@@ -0,0 +1,38 @@
+import argparse
+import json
+import sys
+
+import yaml
+from implicitdict import ImplicitDict
+
+from monitoring.uss_qualifier.resources.files import ExternalFile, load_dict
+from monitoring.uss_qualifier.resources.interuss.geospatial_map import (
+    FeatureCheckTableResource,
+)
+from monitoring.uss_qualifier.resources.interuss.geospatial_map.feature_check_table import (
+    FeatureCheckTableSpecification,
+)
+
+
+def parse_args(argv: list[str]):
+    parser = argparse.ArgumentParser(description="Generate a FeatureCheckTable ")
+    parser.add_argument(
+        "--config-path",
+        action="store",
+        dest="config_path",
+        type=str,
+        help="Path to dict-like file containing a FeatureCheckTableSpecification (may have an anchor suffix like #/foo/bar)",
+    )
+    return parser.parse_args(argv)
+
+
+def generate_table(path: str):
+    spec_dict = load_dict(ExternalFile(path=path))
+    spec = ImplicitDict.parse(spec_dict, FeatureCheckTableSpecification)
+    resource = FeatureCheckTableResource(specification=spec, resource_origin="Script")
+    print(yaml.dump(json.loads(json.dumps(resource.table)), indent=2))
+
+
+if __name__ == "__main__":
+    args = parse_args(sys.argv[1:])
+    generate_table(args.config_path)

monitoring/uss_qualifier/scenarios/interuss/geospatial_map/geospatial_feature_comprehension.py

@@ -98,6 +98,10 @@ def _map_query(self, times: dict[TimeDuringTest, Time]):
                 checks=[query_check, check],
             )
             self.begin_dynamic_test_step(doc)
+            if row.description:
+                self.record_note(
+                    f"{row.geospatial_check_id}.description", row.description
+                )
 
             # Populate filter set
             filter_set = GeospatialFeatureFilter()

monitoring/uss_qualifier/test_data/aus/incident_translator.jsonnet

@@ -0,0 +1,76 @@
+// Translates "incidents" GeoJSON provided via "geojson_example" external variable into a FeatureCheckTable
+local features = std.extVar("geojson_example")["incidents"]["features"];
+local timestamp_of = std.native("timestamp_of");
+local now = timestamp_of("2025-10-11T13:30:08.503557+09:30");
+local no_older_than = 24 * 60 * 60;
+
+local make_row = function(index, feature, volumes)
+  if feature["properties"]["_status"] == "closed" then
+    null
+  else if now - timestamp_of(feature["properties"]["_lastupdate"]) > no_older_than then
+    null
+  else
+    {
+      geospatial_check_id: "TEST_" + std.format("%03d", index + 1),
+      requirement_ids: ["REQ_002"],
+      description: feature["properties"]["_category"] + " at " + feature["properties"]["_datenotified"],
+      operation_rule_set: "Rules1",
+      restriction_source: "ThisRegulator",
+      expected_result: "Block",
+      volumes: volumes
+    };
+
+local make_4d_volume = function(footprint)
+  footprint + {
+    altitude_lower: {value: 0, units: "M", reference: "SFC"},
+    altitude_upper: {value: 100, units: "M", reference: "SFC"},
+    start_time: {time_during_test: "StartOfTestRun"},
+    end_time: {
+      offset_from: {
+        starting_from: {time_during_test: "StartOfTestRun"},
+        offset: "1h"
+      }
+    }
+  };
+
+local make_point_row = function(index, feature)
+  local footprint = {
+    outline_circle: {
+      center: {
+        lat: feature["geometry"]["coordinates"][1],
+        lng: feature["geometry"]["coordinates"][0]
+      },
+      radius: {value: 10, units: "M"}
+    },
+  };
+  local volumes = [make_4d_volume(footprint)];
+  make_row(index, feature, volumes);
+
+local make_polygon_row = function(index, feature)
+  local footprint = {
+    outline_polygon: {
+      vertices: [
+        {lng: coords[0], lat: coords[1]}
+        for coords in feature["geometry"]["coordinates"][0]
+      ]
+    }
+  };
+  local volumes = [make_4d_volume(footprint)];
+  make_row(index, feature, volumes);
+
+local append_row = function(rows, feature)
+  local new_row =
+    if feature["geometry"]["type"] == "Point" then
+      make_point_row(std.length(rows), feature)
+    else if feature["geometry"]["type"] == "Polygon" then
+      make_polygon_row(std.length(rows), feature)
+    else
+      null;
+  if new_row == null then
+    rows
+  else
+    rows + [new_row];
+
+{
+  rows: std.foldl(append_row, features, [])
+}