From 5ba79662917127c3cc1761389068f44302d9aced Mon Sep 17 00:00:00 2001 From: Ethan Yu Date: Wed, 27 May 2026 05:51:45 -0100 Subject: [PATCH] Type UAV schema registry --- .../src/uav/uav/runtime/mission_spec.py | 32 +++++++++++-------- .../src/uav/uav/runtime/schema_generator.py | 20 +++++++----- .../src/uav/uav/runtime/schema_registry.py | 8 ++--- 3 files changed, 35 insertions(+), 25 deletions(-) diff --git a/controls/sae_2025_ws/src/uav/uav/runtime/mission_spec.py b/controls/sae_2025_ws/src/uav/uav/runtime/mission_spec.py index 7a9636db3..69a946784 100644 --- a/controls/sae_2025_ws/src/uav/uav/runtime/mission_spec.py +++ b/controls/sae_2025_ws/src/uav/uav/runtime/mission_spec.py @@ -10,25 +10,31 @@ from .mode_paths import normalize_public_mode_id -try: - from ament_index_python.packages import ( - PackageNotFoundError, - get_package_share_directory, - ) -except ModuleNotFoundError: - - class PackageNotFoundError(Exception): - pass - - def get_package_share_directory(_package_name: str) -> str: - raise PackageNotFoundError - VALID_MISSION_TARGETS = {"uav", "payload"} _TOP_LEVEL_KEYS = {"modes"} _MODE_KEYS = {"mode", "params", "transitions"} +class PackageNotFoundError(Exception): + pass + + +def get_package_share_directory(package_name: str) -> str: + try: + from ament_index_python.packages import ( + PackageNotFoundError as AmentPackageNotFoundError, + get_package_share_directory as ament_get_package_share_directory, + ) + except ModuleNotFoundError as exc: + raise PackageNotFoundError from exc + + try: + return ament_get_package_share_directory(package_name) + except AmentPackageNotFoundError as exc: + raise PackageNotFoundError from exc + + class MissionModeDocumentModel(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) diff --git a/controls/sae_2025_ws/src/uav/uav/runtime/schema_generator.py b/controls/sae_2025_ws/src/uav/uav/runtime/schema_generator.py index ad6ae7c98..04d610d69 100644 --- a/controls/sae_2025_ws/src/uav/uav/runtime/schema_generator.py +++ b/controls/sae_2025_ws/src/uav/uav/runtime/schema_generator.py @@ -11,7 +11,7 @@ import pkgutil from pathlib import Path import types -from typing import Any, Literal, Union, get_args, get_origin, get_type_hints +from typing import Any, Literal, Union, cast, get_args, get_origin, get_type_hints from pydantic import BaseModel, ConfigDict, create_model @@ -30,6 +30,8 @@ registry_path, ) +DefaultKind = Literal["missing", "none", "value", "nan"] + def _doc_summary(obj: object) -> str: doc = inspect.getdoc(obj) or "" @@ -88,8 +90,9 @@ def _normalized_annotation( ) if annotation in {dict, list, tuple, set}: + annotation_name = getattr(annotation, "__name__", repr(annotation)) raise TypeError( - f"Mode '{mode_id_for(mode_class)}' parameter '{name}' must use a typed collection annotation, not bare '{annotation.__name__}'." + f"Mode '{mode_id_for(mode_class)}' parameter '{name}' must use a typed collection annotation, not bare '{annotation_name}'." ) origin = get_origin(annotation) @@ -129,7 +132,7 @@ def mode_params_model(mode_class: type[Mode]) -> type[BaseModel]: f"{mode_class.__name__}Params", __config__=ConfigDict(extra="forbid"), __module__=mode_class.__module__, - **field_definitions, + **cast(Any, field_definitions), ) @@ -158,7 +161,7 @@ def _sanitize_json_schema(value: Any) -> Any: return value -def _default_payload(default: object) -> tuple[str, Any | None]: +def _default_payload(default: object) -> tuple[DefaultKind, Any | None]: if default is inspect.Parameter.empty: return "missing", None if default is None: @@ -181,9 +184,10 @@ def _normalize_annotation_spec(annotation: object) -> dict[str, Any]: return {"kind": "none"} if _is_typed_dict(annotation): - type_hints = get_type_hints(annotation) - required_keys = set(annotation.__required_keys__) - optional_keys = set(annotation.__optional_keys__) + typed_dict = cast(Any, annotation) + type_hints = get_type_hints(typed_dict) + required_keys = set(typed_dict.__required_keys__) + optional_keys = set(typed_dict.__optional_keys__) fields = [] for name in sorted(required_keys | optional_keys): fields.append( @@ -195,7 +199,7 @@ def _normalize_annotation_spec(annotation: object) -> dict[str, Any]: ) return { "kind": "typed_dict", - "name": annotation.__name__, + "name": getattr(typed_dict, "__name__", str(annotation)), "fields": fields, } diff --git a/controls/sae_2025_ws/src/uav/uav/runtime/schema_registry.py b/controls/sae_2025_ws/src/uav/uav/runtime/schema_registry.py index c7d901d10..7e228b5ff 100644 --- a/controls/sae_2025_ws/src/uav/uav/runtime/schema_registry.py +++ b/controls/sae_2025_ws/src/uav/uav/runtime/schema_registry.py @@ -4,7 +4,7 @@ import json import operator from pathlib import Path -from typing import Any, Literal +from typing import Any, Literal, cast from pydantic import BaseModel, ConfigDict, create_model from typing_extensions import NotRequired, Required, TypedDict @@ -110,7 +110,7 @@ def _annotation_from_spec(annotation: dict[str, Any], *, name_hint: str) -> obje return type(None) if kind == "literal": choices = tuple(annotation.get("choices", ())) - return Literal.__getitem__(choices) + return cast(Any, Literal).__getitem__(choices) if kind == "list": return list[ _annotation_from_spec(annotation["item"], name_hint=f"{name_hint}_item") @@ -141,7 +141,7 @@ def _annotation_from_spec(annotation: dict[str, Any], *, name_hint: str) -> obje fields[field["name"]] = Required[field_type] else: fields[field["name"]] = NotRequired[field_type] - return TypedDict(_typed_dict_name(name_hint), fields) + return cast(Any, TypedDict)(_typed_dict_name(name_hint), fields) if kind == "union": options = [ _annotation_from_spec(option, name_hint=f"{name_hint}_{index}") @@ -216,5 +216,5 @@ def params_model_for_entry(mode_id: str) -> type[BaseModel]: f"{entry.class_name}Params", __config__=ConfigDict(extra="forbid"), __module__=__name__, - **field_definitions, + **cast(Any, field_definitions), )