[tooling] Use optional in ImplicitDicts (#1314)

Author: the-glu

monitoring/deployment_manager/systems/configuration.py

@@ -1,4 +1,4 @@
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 
 from monitoring.deployment_manager.systems.dss.configuration import DSS
 from monitoring.deployment_manager.systems.test.configuration import Test
@@ -14,11 +14,11 @@ class KubernetesCluster(ImplicitDict):
 
 
 class DeploymentSpec(ImplicitDict):
-    cluster: KubernetesCluster | None
+    cluster: Optional[KubernetesCluster]
     """Definition of Kubernetes cluster containing this deployment."""
 
-    test: Test | None
+    test: Optional[Test]
     """Test systems in this deployment."""
 
-    dss: DSS | None
+    dss: Optional[DSS]
     """DSS instance in this deployment."""

monitoring/deployment_manager/systems/test/configuration.py

@@ -1,9 +1,9 @@
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 
 
 class TestV1(ImplicitDict):
     namespace: str = "test"
 
 
 class Test(ImplicitDict):
-    v1: TestV1 | None
+    v1: Optional[TestV1]

monitoring/mock_uss/database.py

@@ -1,14 +1,19 @@
 import json
 
-from implicitdict import ImplicitDict, StringBasedDateTime, StringBasedTimeDelta
+from implicitdict import (
+    ImplicitDict,
+    Optional,
+    StringBasedDateTime,
+    StringBasedTimeDelta,
+)
 
 from monitoring.monitorlib.errors import stacktrace_string
 from monitoring.monitorlib.multiprocessing import SynchronizedValue
 
 
 class PeriodicTaskStatus(ImplicitDict):
-    last_execution_time: StringBasedDateTime | None = None
-    period: StringBasedTimeDelta | None = None
+    last_execution_time: Optional[StringBasedDateTime] = None
+    period: Optional[StringBasedTimeDelta] = None
     executing: bool = False
 
 
@@ -43,7 +48,7 @@ class Database(ImplicitDict):
     periodic_tasks: dict[str, PeriodicTaskStatus]
     """Tasks to perform periodically, by name"""
 
-    most_recent_periodic_check: StringBasedDateTime | None
+    most_recent_periodic_check: Optional[StringBasedDateTime]
     """Timestamp of most recent time periodic task loop iterated"""
 
 

monitoring/mock_uss/flights/database.py

@@ -1,7 +1,7 @@
 import json
 from datetime import timedelta
 
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 from uas_standards.astm.f3548.v21.api import OperationalIntent
 
 from monitoring.mock_uss.user_interactions.notifications import UserNotification
@@ -19,7 +19,7 @@ class FlightRecord(ImplicitDict):
 
     flight_info: FlightInfo
     op_intent: OperationalIntent
-    mod_op_sharing_behavior: MockUssFlightBehavior | None = None
+    mod_op_sharing_behavior: Optional[MockUssFlightBehavior] = None
     locked: bool = False
 
 

monitoring/mock_uss/geoawareness/database.py

@@ -1,6 +1,6 @@
 import json
 
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 from uas_standards.eurocae_ed269 import ED269Schema
 from uas_standards.interuss.automated_testing.geo_awareness.v1.api import (
     CreateGeozoneSourceRequest,
@@ -17,8 +17,8 @@ class ExistingRecordException(ValueError):
 class SourceRecord(ImplicitDict):
     definition: CreateGeozoneSourceRequest
     state: GeozoneSourceResponseResult
-    message: str | None
-    geozone_ed269: ED269Schema | None
+    message: Optional[str]
+    geozone_ed269: Optional[ED269Schema]
 
 
 class Database(ImplicitDict):

monitoring/mock_uss/riddp/behavior.py

@@ -1,8 +1,8 @@
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 
 ServiceProviderID = str
 
 
 class DisplayProviderBehavior(ImplicitDict):
-    always_omit_recent_paths: bool | None = False
-    do_not_display_flights_from: list[ServiceProviderID] | None = []
+    always_omit_recent_paths: Optional[bool] = False
+    do_not_display_flights_from: Optional[list[ServiceProviderID]] = []

monitoring/mock_uss/ridsp/behavior.py

@@ -1,4 +1,4 @@
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 from uas_standards.astm.f3411.v19.api import RIDFlight
 
 from monitoring.monitorlib.delay import sleep
@@ -8,10 +8,10 @@
 
 
 class ServiceProviderBehavior(ImplicitDict):
-    switch_latitude_and_longitude_when_reporting: bool | None = False
-    use_agl_instead_of_wgs84_for_altitude: bool | None = False
-    use_feet_instead_of_meters_for_altitude: bool | None = False
-    delay_flight_report_s: int | None = 0
+    switch_latitude_and_longitude_when_reporting: Optional[bool] = False
+    use_agl_instead_of_wgs84_for_altitude: Optional[bool] = False
+    use_feet_instead_of_meters_for_altitude: Optional[bool] = False
+    delay_flight_report_s: Optional[int] = 0
 
 
 def adjust_reported_flight(

monitoring/mock_uss/ridsp/database.py

@@ -1,6 +1,6 @@
 import json
 
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 
 from monitoring.monitorlib.multiprocessing import SynchronizedValue
 from monitoring.monitorlib.rid_automated_testing import injection_api
@@ -14,7 +14,7 @@ class TestRecord(ImplicitDict):
 
     version: str
     flights: list[injection_api.TestFlight]
-    isa_version: str | None = None
+    isa_version: Optional[str] = None
 
     def __init__(self, **kwargs):
         kwargs["flights"] = [

monitoring/mock_uss/tracer/log_types.py

@@ -3,7 +3,7 @@
 import sys
 from abc import abstractmethod
 
-from implicitdict import ImplicitDict, StringBasedDateTime
+from implicitdict import ImplicitDict, Optional, StringBasedDateTime
 
 from monitoring.monitorlib.fetch import RequestDescription, summarize
 from monitoring.monitorlib.fetch import rid as rid_fetch
@@ -124,7 +124,7 @@ def prefix_code() -> str:
     existing_subscription: rid_fetch.FetchedSubscription
     """Subscription, as read from the DSS just before deletion."""
 
-    deleted_subscription: rid_mutate.ChangedSubscription | None
+    deleted_subscription: Optional[rid_mutate.ChangedSubscription]
     """Subscription returned from DSS upon deletion."""
 
 
@@ -214,7 +214,7 @@ def prefix_code() -> str:
     existing_subscription: scd_fetch.FetchedSubscription
     """Subscription, as read from the DSS just before deletion."""
 
-    deleted_subscription: scd_mutate.MutatedSubscription | None
+    deleted_subscription: Optional[scd_mutate.MutatedSubscription]
     """Subscription returned from DSS upon deletion."""
 
 
@@ -270,7 +270,7 @@ class ObservationAreaImportError(TracerLogEntry):
     def prefix_code() -> str:
         return "import_obs_areas_error"
 
-    rid_subscriptions: rid_fetch.FetchedSubscriptions | None
+    rid_subscriptions: Optional[rid_fetch.FetchedSubscriptions]
     """Result of attempting to fetch RID subscriptions"""
 
 

monitoring/mock_uss/tracer/observation_areas.py

@@ -1,4 +1,4 @@
-from implicitdict import ImplicitDict
+from implicitdict import ImplicitDict, Optional
 
 from monitoring.monitorlib.geotemporal import Volume4D
 from monitoring.monitorlib.infrastructure import AuthSpec
@@ -45,7 +45,7 @@ class F3548ObservationArea(ImplicitDict):
     poll: bool
     """This area observes by periodically polling for information."""
 
-    subscription_id: str | None = None
+    subscription_id: Optional[str] = None
     """The F3548 subscription ID established to provide observation via notifications."""
 
 
@@ -58,10 +58,10 @@ class ObservationArea(ImplicitDict):
     area: Volume4D
     """Spatial-temporal area being observed."""
 
-    f3411: F3411ObservationArea | None = None
+    f3411: Optional[F3411ObservationArea] = None
     """How F3411 information is being observed (or not observed, if not specified)."""
 
-    f3548: F3548ObservationArea | None = None
+    f3548: Optional[F3548ObservationArea] = None
     """How F3548 information is being observed (or not observed, if not specified)."""
 
     @property
@@ -75,12 +75,12 @@ def polls(self) -> bool:
 class F3411ObservationAreaRequest(ImplicitDict):
     """How to observe F3411 activity."""
 
-    auth_spec: AuthSpec | None = None
+    auth_spec: Optional[AuthSpec] = None
     """If specified, use this auth spec when performing observation activities.
 
     If not specified or blank, use auth spec provided on the command line."""
 
-    dss_base_url: str | None = None
+    dss_base_url: Optional[str] = None
     """If specified, use the DSS at this base URL when performing relevant observation activities.
 
     If not specified or blank, use DSS URL provided on the command line."""
@@ -98,12 +98,12 @@ class F3411ObservationAreaRequest(ImplicitDict):
 class F3548ObservationAreaRequest(ImplicitDict):
     """How to observe F3548 activity."""
 
-    auth_spec: AuthSpec | None = None
+    auth_spec: Optional[AuthSpec] = None
     """If specified, use this auth spec when performing observation activities.
 
     If not specified or blank, use auth spec provided on the command line."""
 
-    dss_base_url: str | None = None
+    dss_base_url: Optional[str] = None
     """If specified, use the DSS at this base URL when performing relevant observation activities.
 
     If not specified or blank, use DSS URL provided on the command line."""
@@ -127,10 +127,10 @@ class ObservationAreaRequest(ImplicitDict):
     area: Volume4D
     """Spatial-temporal area that should be observed."""
 
-    f3411: F3411ObservationAreaRequest | None = None
+    f3411: Optional[F3411ObservationAreaRequest] = None
     """How to observe F3411 (NetRID) activity."""
 
-    f3548: F3548ObservationAreaRequest | None = None
+    f3548: Optional[F3548ObservationAreaRequest] = None
     """How to observe F3548 (strategic coordination, conformance monitoring, and constraints) activity."""
 
     @property
@@ -168,7 +168,7 @@ class ImportObservationAreasRequest(ImplicitDict):
     area: Volume4D
     """Spatial-temporal area containing subscriptions to be imported."""
 
-    f3411: RIDVersion | None = None
+    f3411: Optional[RIDVersion] = None
     """If specified, search for subscriptions using this F3411 version."""
 
     f3548: bool = False