[benchmarker] Add SCD subscription actions

monitoring/benchmarker/configurations/actions/action.py

@@ -3,6 +3,7 @@
 from implicitdict import ImplicitDict
 
 from monitoring.benchmarker.configurations.actions.f3411 import F3411ActionSpecification
+from monitoring.benchmarker.configurations.actions.f3548 import F3548ActionSpecification
 from monitoring.benchmarker.configurations.artifacts.artifact import (
     ArtifactSpecification,
 )
@@ -52,4 +53,6 @@ class BenchmarkActionSpecification(ImplicitDict):
 
     f3411: Optional[F3411ActionSpecification]
 
+    f3548: Optional[F3548ActionSpecification]
+
     generate_artifacts: Optional[GenerateArtifactsActionSpecification]

monitoring/benchmarker/configurations/actions/f3548.py

@@ -0,0 +1,58 @@
+from typing import Optional
+
+from implicitdict import ImplicitDict, StringBasedTimeDelta
+from uas_standards.astm.f3548.v21.api import SubscriptionID
+
+from monitoring.benchmarker.configurations.actions.astm import (
+    SubscriptionCreationMode,
+    SubscriptionDeletionMode,
+)
+from monitoring.monitorlib.geo import Altitude, LatLngBoundingBox
+
+
+class Subscription(ImplicitDict):
+    subscription_id: SubscriptionID
+    """ID of the single subscription to create."""
+
+    duration: StringBasedTimeDelta
+    """Duration of the subscription, from the time it is created."""
+
+    area: LatLngBoundingBox
+    """Horizontal area this subscription should cover."""
+
+    min_alt: Altitude
+    """Altitude below which this subscription should not apply."""
+
+    max_alt: Altitude
+    """Altitude above which this subscription should not apply."""
+
+    notify_for_op_intents: Optional[bool]
+    """Whether to receive notifications for operational intents. Defaults to True if not specified."""
+
+    notify_for_constraints: Optional[bool]
+    """Whether to receive notifications for constraints. Defaults to False if not specified."""
+
+
+class CreateSubscription(ImplicitDict):
+    """Create a subscription."""
+
+    subscription: Subscription
+    """Characteristics of subscription to create."""
+
+    mode: SubscriptionCreationMode
+    """Desired creation behavior."""
+
+
+class DeleteSubscription(ImplicitDict):
+    subscription_id: SubscriptionID
+    """ID of the subscription to delete."""
+
+    mode: SubscriptionDeletionMode
+    """Desired deletion behavior."""
+
+
+class F3548ActionSpecification(ImplicitDict):
+    """Actions pertaining to ASTM F3548 SCD."""
+
+    create_subscription: Optional[CreateSubscription]
+    delete_subscription: Optional[DeleteSubscription]

monitoring/benchmarker/configurations/interuss/scd/single_s2_cell.jsonnet

@@ -1,6 +1,7 @@
 local test_name = 'Single S2 cell';
 local num_uss = 3;
 local num_nodes = 3;
+local num_subscriptions = 8;
 local dss_config_names = ['Existing local DSS deployment'];
 local users_per_step = 3;
 
@@ -72,6 +73,37 @@ local shape = {
         defined_artifact_indices: [0, 1],
       },
     },
+  ] + [
+    {
+      name: 'Create subscription %d' % sub_index,
+      f3548: {
+        create_subscription: {
+          subscription: {
+            subscription_id: '16b87239-6063-47d4-a2ff-%d05086859f32' % (sub_index - 1),
+            duration: '23h',
+            area: {
+              lat_min: 34 - 0.00001,
+              lng_min: -118 - 0.00001,
+              lat_max: 34 + 0.00001,
+              lng_max: -118 + 0.00001,
+            },
+            min_alt: {value: 0, units: 'M', reference: 'W84'},
+            max_alt: {value: 3000, units: 'M', reference: 'W84'},
+          },
+          mode: 'GetDeleteCreate',
+        },
+      },
+    } for sub_index in std.range(1, num_subscriptions)
+  ] + [
+    {
+      name: 'Delete subscription %d' % sub_index,
+      f3548: {
+        delete_subscription: {
+          subscription_id: '16b87239-6063-47d4-a2ff-%d05086859f32' % (sub_index - 1),
+          mode: 'GetDeleteIfExist',
+        },
+      },
+    } for sub_index in std.range(1, num_subscriptions)
   ],
 
   user_types: [
@@ -197,8 +229,11 @@ local shape = {
     [
       {
         name: '%s: %s for USS %d' % [dss_config_names[dss_config - 1], test_name,  uss],
+        [if uss == 1 then "setup"]: ['Create subscription %d' % sub_index for sub_index in std.range(1, num_subscriptions)],
         load: 'Flight planner ramp for USS %d' % uss,
-        [if uss < num_uss || dss_config < std.length(dss_config_names) then "teardown"]: ['Generate intermediate artifacts'],
+        [if uss < num_uss || dss_config < std.length(dss_config_names) || uss == num_uss then "teardown"]:
+          (if uss < num_uss || dss_config < std.length(dss_config_names) then ['Generate intermediate artifacts'] else [])
+          + (if uss == num_uss then ['Delete subscription %d' % sub_index for sub_index in std.range(1, num_subscriptions)] else []),
       } for uss in std.range(1, num_uss)
     ] for dss_config in std.range(1, std.length(dss_config_names))
   ]),

monitoring/benchmarker/engine/actions/actions.py

@@ -6,6 +6,7 @@
 )
 from monitoring.benchmarker.configurations.configuration import BenchmarkConfiguration
 from monitoring.benchmarker.engine.actions.f3411 import run_f3411_action
+from monitoring.benchmarker.engine.actions.f3548 import run_f3548_action
 from monitoring.benchmarker.engine.actions.generate_artifacts import (
     generate_intermediate_artifacts,
 )
@@ -63,6 +64,13 @@ def run_scenario_actions(
                 )
             run_f3411_action(action_name, action_spec.f3411, resource_pool)
             action_performed = True
+        if "f3548" in action_spec and action_spec.f3548 is not None:
+            if resource_pool is None:
+                raise ValueError(
+                    f"Resource pool is required to execute action '{action_name}' with F3548 specification"
+                )
+            run_f3548_action(action_name, action_spec.f3548, resource_pool)
+            action_performed = True
 
         if not action_performed:
             raise NotImplementedError(

monitoring/benchmarker/engine/actions/f3548.py

@@ -0,0 +1,187 @@
+from datetime import UTC, datetime
+from typing import Any
+
+from loguru import logger
+from uas_standards.astm.f3548.v21.constants import Scope
+
+from monitoring.benchmarker.configurations.actions.action import BenchmarkActionName
+from monitoring.benchmarker.configurations.actions.astm import (
+    SubscriptionCreationMode,
+    SubscriptionDeletionMode,
+)
+from monitoring.benchmarker.configurations.actions.f3548 import (
+    CreateSubscription,
+    DeleteSubscription,
+    F3548ActionSpecification,
+)
+from monitoring.monitorlib.testing import make_fake_url
+from monitoring.uss_qualifier.resources.astm.f3548.v21.dss import (
+    DSSInstance,
+    DSSInstanceResource,
+    DSSInstancesResource,
+)
+from monitoring.uss_qualifier.resources.definitions import ResourceID
+
+
+def get_dss_instances(resource_pool: dict[ResourceID, Any]) -> list[DSSInstance]:
+    """Retrieve all F3548 DSS instances from the resource pool."""
+    scopes_required = {
+        Scope.StrategicCoordination.value: "managing subscriptions for strategic conflict detection",
+    }
+    dss_instances: list[DSSInstance] = []
+    for res in resource_pool.values():
+        if isinstance(res, DSSInstanceResource):
+            dss_instances.append(res.get_instance(scopes_required))
+        elif isinstance(res, DSSInstancesResource):
+            for dss_instance_res in res.dss_instances:
+                dss_instances.append(dss_instance_res.get_instance(scopes_required))
+    return dss_instances
+
+
+def create_subscription(
+    spec: CreateSubscription,
+    resource_pool: dict[ResourceID, Any],
+) -> None:
+    dss_instances = get_dss_instances(resource_pool)
+    if not dss_instances:
+        raise ValueError("No ASTM F3548 DSS instances found in resource pool")
+    sub = spec.subscription
+    dss_instance = dss_instances[0]
+
+    if spec.mode == SubscriptionCreationMode.GetDeleteCreate:
+        logger.info(
+            f"F3548 Action: Checking if subscription '{sub.subscription_id}' exists before creating..."
+        )
+        fetched_sub = dss_instance.get_subscription(sub.subscription_id)
+        if fetched_sub.status_code == 200 and fetched_sub.subscription is not None:
+            logger.info(
+                f"F3548 Action: Existing subscription '{sub.subscription_id}' found (version {fetched_sub.subscription.version}); deleting it..."
+            )
+            del_result = dss_instance.delete_subscription(
+                sub_id=sub.subscription_id,
+                sub_version=fetched_sub.subscription.version,
+            )
+            if not del_result.success:
+                raise RuntimeError(
+                    f"Failed to delete existing subscription '{sub.subscription_id}' during GetDeleteCreate: {del_result.errors}"
+                )
+        elif fetched_sub.status_code == 404:
+            logger.info(
+                f"F3548 Action: Subscription '{sub.subscription_id}' does not exist; proceeding to create."
+            )
+        else:
+            raise RuntimeError(
+                f"Failed to query subscription '{sub.subscription_id}' during GetDeleteCreate: {fetched_sub.errors}"
+            )
+
+        logger.info(f"F3548 Action: Creating subscription '{sub.subscription_id}'...")
+        uss_base_url = make_fake_url()
+        t0 = datetime.now(UTC)
+        notify_for_op_intents = (
+            sub.notify_for_op_intents
+            if "notify_for_op_intents" in sub and sub.notify_for_op_intents is not None
+            else True
+        )
+        notify_for_constraints = (
+            sub.notify_for_constraints
+            if "notify_for_constraints" in sub
+            and sub.notify_for_constraints is not None
+            else False
+        )
+        create_result = dss_instance.upsert_subscription(
+            area_vertices=sub.area.to_latlngrect(),
+            start_time=t0,
+            end_time=t0 + sub.duration.timedelta,
+            base_url=uss_base_url,
+            sub_id=sub.subscription_id,
+            notify_for_op_intents=notify_for_op_intents,
+            notify_for_constraints=notify_for_constraints,
+            min_alt_m=sub.min_alt.to_w84_m(),
+            max_alt_m=sub.max_alt.to_w84_m(),
+        )
+        if not create_result.success:
+            raise RuntimeError(
+                f"Failed to create subscription '{sub.subscription_id}': {create_result.errors}"
+            )
+        logger.info(
+            f"F3548 Action: Successfully created subscription '{sub.subscription_id}'."
+        )
+    else:
+        raise NotImplementedError(
+            f"Unsupported subscription creation mode '{spec.mode}'"
+        )
+
+
+def delete_subscription(
+    spec: DeleteSubscription,
+    resource_pool: dict[ResourceID, Any],
+) -> None:
+    dss_instances = get_dss_instances(resource_pool)
+    if not dss_instances:
+        raise ValueError("No ASTM F3548 DSS instances found in resource pool")
+
+    if spec.mode == SubscriptionDeletionMode.GetDeleteIfExist:
+        logger.info(
+            f"F3548 Action: Checking if subscription '{spec.subscription_id}' exists before deleting..."
+        )
+        deleted = False
+        for dss_instance in dss_instances:
+            fetched_sub = dss_instance.get_subscription(spec.subscription_id)
+            if fetched_sub.status_code == 200 and fetched_sub.subscription is not None:
+                logger.info(
+                    f"F3548 Action: Existing subscription '{spec.subscription_id}' found (version {fetched_sub.subscription.version}); deleting it..."
+                )
+                del_result = dss_instance.delete_subscription(
+                    sub_id=spec.subscription_id,
+                    sub_version=fetched_sub.subscription.version,
+                )
+                if not del_result.success:
+                    raise RuntimeError(
+                        f"Failed to delete subscription '{spec.subscription_id}': {del_result.errors}"
+                    )
+                logger.info(
+                    f"F3548 Action: Successfully deleted subscription '{spec.subscription_id}'."
+                )
+                deleted = True
+                break
+            elif fetched_sub.status_code == 404:
+                continue
+            else:
+                raise RuntimeError(
+                    f"Failed to query subscription '{spec.subscription_id}' during GetDeleteIfExist: {fetched_sub.errors}"
+                )
+
+        if not deleted:
+            logger.info(
+                f"F3548 Action: Subscription '{spec.subscription_id}' did not exist; nothing to delete."
+            )
+    else:
+        raise NotImplementedError(
+            f"Unsupported subscription deletion mode '{spec.mode}'"
+        )
+
+
+def run_f3548_action(
+    action_name: BenchmarkActionName,
+    f3548_spec: F3548ActionSpecification,
+    resource_pool: dict[ResourceID, Any],
+) -> None:
+    action_performed = False
+    if (
+        "create_subscription" in f3548_spec
+        and f3548_spec.create_subscription is not None
+    ):
+        logger.info(f"Action '{action_name}': Creating F3548 subscription...")
+        create_subscription(f3548_spec.create_subscription, resource_pool)
+        action_performed = True
+    if (
+        "delete_subscription" in f3548_spec
+        and f3548_spec.delete_subscription is not None
+    ):
+        logger.info(f"Action '{action_name}': Deleting F3548 subscription...")
+        delete_subscription(f3548_spec.delete_subscription, resource_pool)
+        action_performed = True
+    if not action_performed:
+        raise ValueError(
+            f"Action '{action_name}' F3548ActionSpecification did not specify any supported action"
+        )

schemas/monitoring/benchmarker/configurations/actions/action/BenchmarkActionSpecification.json

@@ -17,6 +17,16 @@
         }
       ]
     },
+    "f3548": {
+      "oneOf": [
+        {
+          "type": "null"
+        },
+        {
+          "$ref": "../f3548/F3548ActionSpecification.json"
+        }
+      ]
+    },
     "generate_artifacts": {
       "oneOf": [
         {

schemas/monitoring/benchmarker/configurations/actions/f3548/CreateSubscription.json

@@ -0,0 +1,27 @@
+{
+  "$id": "https://github.com/interuss/monitoring/blob/main/schemas/monitoring/benchmarker/configurations/actions/f3548/CreateSubscription.json",
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "description": "Create a subscription.\n\nmonitoring.benchmarker.configurations.actions.f3548.CreateSubscription, as defined in monitoring/benchmarker/configurations/actions/f3548.py",
+  "properties": {
+    "$ref": {
+      "description": "Path to content that replaces the $ref",
+      "type": "string"
+    },
+    "mode": {
+      "description": "Desired creation behavior.",
+      "enum": [
+        "GetDeleteCreate"
+      ],
+      "type": "string"
+    },
+    "subscription": {
+      "$ref": "Subscription.json",
+      "description": "Characteristics of subscription to create."
+    }
+  },
+  "required": [
+    "mode",
+    "subscription"
+  ],
+  "type": "object"
+}