Skip to content

[mock_uss] Set of typing fixes #1199

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Oct 8, 2025
Merged

[mock_uss] Set of typing fixes #1199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
684bd9c
[mock_uss] Set of typing fixes
the-glu Oct 1, 2025
e19a5f0
Merge branch 'main' into octobrer_typing
the-glu Oct 2, 2025
9ebf05e
Merge branch 'main' into octobrer_typing
the-glu Oct 6, 2025
ec5b73c
Updates from #1199's comments
the-glu Oct 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
770 changes: 6 additions & 764 deletions .basedpyright/baseline.json
View file
Open in desktop

Large diffs are not rendered by default.

24 changes: 13 additions & 11 deletions monitoring/mock_uss/tracer/context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
KEY_TRACER_OUTPUT_FOLDER,
)
from monitoring.mock_uss.tracer.observation_areas import ObservationAreaID
from monitoring.mock_uss.tracer.tracerlog import Logger
from monitoring.mock_uss.tracer.tracerlog import DummyLogger, Logger
from monitoring.monitorlib import infrastructure
from monitoring.monitorlib.auth import make_auth_adapter
from monitoring.monitorlib.fetch import scd
Expand All @@ -23,7 +23,7 @@
scd_cache: dict[ObservationAreaID, dict[str, scd.FetchedEntity]] = {}


def _get_tracer_logger() -> Logger | None:
def _get_tracer_logger() -> Logger:
kml_server = webapp.config[KEY_TRACER_KML_SERVER]
kml_folder = webapp.config[KEY_TRACER_KML_FOLDER]
output_folder = webapp.config[KEY_TRACER_OUTPUT_FOLDER]
Expand All @@ -36,7 +36,7 @@ def _get_tracer_logger() -> Logger | None:
if kml_server
else None
)
return Logger(output_folder, kml_session) if output_folder else None
return Logger(output_folder, kml_session) if output_folder else DummyLogger()


tracer_logger: Logger = _get_tracer_logger()
Expand All @@ -57,14 +57,15 @@ def resolve_auth_spec(requested_auth_spec: AuthSpec | None) -> AuthSpec:
return requested_auth_spec


def resolve_rid_dss_base_url(dss_base_url: str, rid_version: RIDVersion) -> str:
def resolve_rid_dss_base_url(dss_base_url: str | None, rid_version: RIDVersion) -> str:
if not dss_base_url:
if KEY_DSS_URL not in webapp.config or not webapp.config[KEY_DSS_URL]:
dss_base_url = webapp.config.get(KEY_DSS_URL)

if not dss_base_url:
raise ValueError(
"DSS base URL was not specified explicitly nor in mock_uss_configuration"
)
else:
dss_base_url = webapp.config[KEY_DSS_URL]

if rid_version == RIDVersion.f3411_19:
return dss_base_url
elif rid_version == RIDVersion.f3411_22a:
Expand All @@ -75,14 +76,15 @@ def resolve_rid_dss_base_url(dss_base_url: str, rid_version: RIDVersion) -> str:
)


def resolve_scd_dss_base_url(dss_base_url: str) -> str:
def resolve_scd_dss_base_url(dss_base_url: str | None) -> str:
if not dss_base_url:
if KEY_DSS_URL not in webapp.config or not webapp.config[KEY_DSS_URL]:
dss_base_url = webapp.config.get(KEY_DSS_URL)

if not dss_base_url:
raise ValueError(
"DSS base URL was not specified explicitly nor in mock_uss_configuration"
)
else:
dss_base_url = webapp.config[KEY_DSS_URL]

return dss_base_url


Expand Down
10 changes: 6 additions & 4 deletions monitoring/mock_uss/tracer/diff.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
from typing import cast

from monitoring.monitorlib import formatting
from monitoring.monitorlib.fetch import rid, scd, summarize


def isa_diff_text(a: rid.FetchedISAs | None, b: rid.FetchedISAs | None) -> str:
"""Create text to display to a real-time user describing a change in ISAs."""
a_summary = summarize.isas(a) if a else {}
a_summary = summarize.limit_long_arrays(a_summary, 6)
a_summary = cast(dict, summarize.limit_long_arrays(a_summary, 6))
b_summary = summarize.isas(b) if b else {}
b_summary = summarize.limit_long_arrays(b_summary, 6)
b_summary = cast(dict, summarize.limit_long_arrays(b_summary, 6))
if b is not None and b.success and a is not None and not a.success:
a_summary = {}
if a is not None and a.success and b is not None and not b.success:
Expand All @@ -26,9 +28,9 @@ def entity_diff_text(
if entity_type and "_" in entity_type:
entity_type = entity_type[0 : entity_type.index("_")]
a_summary = summarize.entities(a, entity_type) if a else {}
a_summary = summarize.limit_long_arrays(a_summary, 6)
a_summary = cast(dict, summarize.limit_long_arrays(a_summary, 6))
b_summary = summarize.entities(b, entity_type) if b else {}
b_summary = summarize.limit_long_arrays(b_summary, 6)
b_summary = cast(dict, summarize.limit_long_arrays(b_summary, 6))
if b is not None and b.success and a is not None and not a.success:
a_summary = {}
if a is not None and a.success and b is not None and not b.success:
Expand Down
60 changes: 44 additions & 16 deletions monitoring/mock_uss/tracer/kml.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,11 @@
import glob
import os
import re
from abc import abstractmethod
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from enum import Enum
from typing import Protocol
from typing import Protocol, TypeVar

import yaml
from implicitdict import ImplicitDict
Expand Down Expand Up @@ -40,7 +41,8 @@ def __enter__(self):
self._start_time = datetime.now(UTC)

def __exit__(self, exc_type, exc_val, exc_tb):
self.elapsed_time += datetime.now(UTC) - self._start_time
if self._start_time:
self.elapsed_time += datetime.now(UTC) - self._start_time


class VolumeType(str, Enum):
Expand All @@ -57,10 +59,14 @@ class HistoricalVolumesCollection:
active_at: datetime


class HistoricalVolumesRenderer(Protocol):
TracerLogEntryType = TypeVar("TracerLogEntryType", bound=TracerLogEntry)


class HistoricalVolumesRenderer[TracerLogEntryType](Protocol):
@abstractmethod
def __call__(
self,
log_entry: TracerLogEntry,
log_entry: TracerLogEntryType,
existing_volume_collections: list[HistoricalVolumesCollection],
) -> list[HistoricalVolumesCollection]:
"""Function that generates named collections of 4D volumes from a tracer log entry.
Expand All @@ -71,6 +77,7 @@ def __call__(

Returns: Collection of 4D volume collections.
"""
raise NotImplementedError


@dataclass
Expand Down Expand Up @@ -126,6 +133,9 @@ def _historical_volumes_op_intent_notification(
existing_volume_collections: list[HistoricalVolumesCollection],
) -> list[HistoricalVolumesCollection]:
try:
if log_entry.request.json is None:
raise ValueError("No json data in log entry")

req = ImplicitDict.parse(
log_entry.request.json, PutOperationalIntentDetailsParameters
)
Expand All @@ -136,7 +146,7 @@ def _historical_volumes_op_intent_notification(
return []
assert isinstance(req, PutOperationalIntentDetailsParameters)

claims = get_token_claims(log_entry.request.headers)
claims = get_token_claims(log_entry.request.headers or {})
manager = claims.get("sub", "[Unknown manager]")
name = f"{manager} {req.operational_intent_id}"

Expand All @@ -148,7 +158,7 @@ def _historical_volumes_op_intent_notification(
else:
version = "[deleted]"
state = "Ended"
volumes = []
volumes = Volume4DCollection()

# See if this op intent version already has a volumes collection
already_defined = False
Expand Down Expand Up @@ -183,6 +193,9 @@ def _historical_volumes_op_intent_poll(
# Add newly-polled operational intents
for op_intent_id, query in log_entry.poll.uss_queries.items():
try:
if query.json_result is None:
raise ValueError("No json result in query")

resp = ImplicitDict.parse(
query.json_result, GetOperationalIntentDetailsResponse
)
Expand Down Expand Up @@ -222,6 +235,9 @@ def _historical_volumes_op_intent_poll(
# Remove any existing operational intents that no longer exist as of this poll
for cached_op_intent_id, cached_query in log_entry.poll.cached_uss_queries.items():
try:
if cached_query.json_result is None:
raise ValueError("No json result in query")

resp = ImplicitDict.parse(
cached_query.json_result, GetOperationalIntentDetailsResponse
)
Expand Down Expand Up @@ -274,26 +290,38 @@ class VolumesFolder:
def truncate(self, latest_time: Time) -> None:
to_remove = []
for v in self.volumes:
if v.volume.time_start.datetime > latest_time.datetime:
if (
v.volume.time_start
and v.volume.time_start.datetime > latest_time.datetime
):
to_remove.append(v)
elif v.volume.time_end.datetime > latest_time.datetime:
elif (
v.volume.time_end and v.volume.time_end.datetime > latest_time.datetime
):
v.volume.time_end = latest_time
for v in to_remove:
self.volumes.remove(v)
for c in self.children:
c.truncate(latest_time)

def to_kml_folder(self) -> kml.Folder:
def dt(t: Time) -> int:
return round((t.datetime - self.reference_time.datetime).total_seconds())

def to_kml_folder(self):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why drop the return type annotation?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because kml.Folder is not a type, it's a variable, they seems to use ElementMaker to generate them on the fly.

Not sure how to do better in that case ^^'

Copy link
Member

@BenjaminPelletier BenjaminPelletier Oct 6, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems like this is an instance where the strictness of the checker might lead us to lower the quality of our codebase to make it happy -- I think we need a clear plan to guard against that happening. pykml seems to generate types at runtime in a way I don't particularly like, but I think we need to be able to deal with unusual approaches like that since we can't afford to write all our own libraries. I can think of a few ways to approach this, though I think probably option 2 is probably the best balance of effort and utility:

  1. Just leave errors we shouldn't correct in the baseline until we've resolved all the other errors. This seems like the easiest approach, but has the downside that people later trying to fix errors will not be able to see that the error has already been evaluated by an earlier reviewer.
  2. Add explicit per-line/occurence opt-out flags. This seems to be what we're doing in general (noqa) and seems not too hard, but also signals later reviewers unlike option 1.
  3. Add something that works technically and contains more information -- like, create our own static Folder typing object that refers to kml and include comments about usage and properties. This seems like too much work :)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do agree with you, but I'm not sure in that particular case that the checker is wrong, see that comment on pyright: microsoft/pyright#8129 (comment)

I'm not sure having an invalid annotation ignored is not worse than having no annotation ;)

One alternative would be to type it with the base class (ObjectifiedElement).

And finally, for that specific function, do we really need the 'correctest' type declared? to_kml_folder is internal to kml.py and no exposed anywhere ^^'

Witch solution do you prefer between opt-out, ObjectifiedElement and implicit typing? I will apply it then everywhere for pykml, that not the first one for that lib that the checker doesn't like ;)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that would be something like option 3, but we wouldn't want just ObjectifiedElement since that contains less information than "KML Folder". I think this is probably highlighting the idea that type annotations are useful for two purposes: automated static type checkers, and also humans (IDE hints, etc). I'm a big fan of all the new benefits we're getting from more extensive automated static type checking, but I want to make sure we don't lose or substantively degrade the benefit humans were receiving from type annotations. Of course this single instance isn't worth the discussion, but I do think it's worth having an idea of what we want to do in general to avoid other instances of losing the semantic coding "this function returns a folder" while still retaining the benefits of more aggressive automated type checking.

I don't think we need to hold this PR up on an optimal outcome, however.

if self.reference_time:
description = f"Relative to {self.reference_time}"
folder = kml.Folder(kml.name(self.name), kml.description(description))
else:
folder = kml.Folder(kml.name(self.name))

for v in self.volumes:
name = f"{v.name} {dt(v.volume.time_start)}s-{dt(v.volume.time_end)}s"
name = v.name

if self.reference_time:
base_time = self.reference_time.datetime

def dt(t: Time) -> int:
return round((t.datetime - base_time).total_seconds())

name = f"{name} {dt(v.volume.time_start) if v.volume.time_start else '?'}s-{dt(v.volume.time_end) if v.volume.time_end else '?'}s"

folder.append(
make_placemark_from_volume(v.volume, name=name, style_url=v.style)
)
Expand Down Expand Up @@ -408,13 +436,13 @@ def render_historical_kml(log_folder: str) -> str:
version_folder.children.append(future_folder)

for i, v in enumerate(hvc.volumes):
if v.time_end.datetime <= hvc.active_at:
if v.time_end and v.time_end.datetime <= hvc.active_at:
# This volume ended before the collection was declared, so it never actually existed
continue
if v.time_start.datetime < hvc.active_at:
if v.time_start and v.time_start.datetime < hvc.active_at:
# Volume is declared in the past, but it's only visible starting now
v.time_start = t_hvc
elif v.time_start.datetime > hvc.active_at:
elif v.time_start and v.time_start.datetime > hvc.active_at:
# Add a "future" volume between when this volume was declared and its start time
future_v = Volume4D(v)
future_v.time_end = v.time_start
Expand Down
8 changes: 6 additions & 2 deletions monitoring/mock_uss/tracer/observation_areas.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,9 @@ class ObservationArea(ImplicitDict):
@property
def polls(self) -> bool:
"""Whether any of the observation activity involves periodic polling."""
return (self.f3411 and self.f3411.poll) or (self.f3548 and self.f3548.poll)
return bool(self.f3411 and self.f3411.poll) or bool(
self.f3548 and self.f3548.poll
)


class F3411ObservationAreaRequest(ImplicitDict):
Expand Down Expand Up @@ -134,7 +136,9 @@ class ObservationAreaRequest(ImplicitDict):
@property
def polls(self) -> bool:
"""Whether any of the observation activity requested involves periodic polling."""
return (self.f3411 and self.f3411.poll) or (self.f3548 and self.f3548.poll)
return bool(self.f3411 and self.f3411.poll) or bool(
self.f3548 and self.f3548.poll
)


class ListObservationAreasResponse(ImplicitDict):
Expand Down
4 changes: 2 additions & 2 deletions monitoring/mock_uss/tracer/routes/observation_areas.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@


@webapp.route("/tracer/observation_areas", methods=["GET"])
@ui_auth.login_required
@ui_auth.login_required()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a reason to change this? login_required is written so that it works with or without parentheses, and adding them with no parameters seems to just add an extra call.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

basedpyright seems to be confused in the non-function mode:

error: Impossible d’affecter l’argument de type « ((f: Unknown) -> _Wrapped[..., Unknown, ..., Unknown | Response]) | _Wrapped[..., Unknown, ..., Unknown | Response] » au paramètre de type « T_route@route »
    Le type « ((f: Unknown) -> _Wrapped[..., Unknown, ..., Unknown | Response]) | _Wrapped[..., Unknown, ..., Unknown | Response] » n’est pas assignable au type « RouteCallable »
      Le type « ((f: Unknown) -> _Wrapped[..., Unknown, ..., Unknown | Response]) | _Wrapped[..., Unknown, ..., Unknown | Response] » n’est pas assignable au type « RouteCallable »
        Le type « (f: Unknown) -> _Wrapped[..., Unknown, ..., Unknown | Response] » n’est pas assignable au type « RouteCallable »
          Le type « (f: Unknown) -> _Wrapped[..., Unknown, ..., Unknown | Response] » n’est pas assignable au type « (...) -> ResponseReturnValue »
            Le type de retour de fonction "_Wrapped[..., Unknown, ..., Unknown | Response]" est incompatible avec le type "ResponseReturnValue"
          Le type « (f: Unknown) -> _Wrapped[..., Unknown, ..., Unknown | Response] » n’est pas assignable au type « (...) -> Awaitable[ResponseReturnValue] »
            Le type de retour de fonction "_Wrapped[..., Unknown, ..., Unknown | Response]" est incompatible avec le type "Awaitable[ResponseReturnValue]" (reportArgumentType)

And I think (but that my opinion ^^') that it look cleaner to always have decorator in one consistent mode instead, meaning we can also write it more cleaning. I'm not sure we gain much by avoid one call (and it's not even sure it's avoided ;) )

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems like some decorators are written to have no parentheses, some are written to have parentheses, and some are written so either is fine. So, I'm not sure how we can necessarily be consistent if we sometimes use decorators from libraries. It worries me a little that the checker seems to falsely flag this as incorrect, but parentheses certainly aren't wrong or worse, so perhaps it's not worth pursuing this particular checker failure further.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, I'm not sure how we can necessarily be consistent if we sometimes use decorators from libraries

Yes, I was thinking about consistently for a single decorator, not across them ^^'

def tracer_list_observation_areas() -> flask.Response:
with db as tx:
result = ListObservationAreasResponse(
Expand Down Expand Up @@ -131,7 +131,7 @@ def tracer_import_observation_areas() -> tuple[str, int] | flask.Response:
elif request.area.volume.outline_polygon:
points = [
s2sphere.LatLng.from_degrees(p.lat, p.lng)
for p in request.area.volume.outline_polygon.vertices
for p in request.area.volume.outline_polygon.vertices or []
]
else:
raise NotImplementedError(
Expand Down
8 changes: 8 additions & 0 deletions monitoring/mock_uss/tracer/routes/scd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,10 @@ def tracer_scd_v21_operation_notification(observation_area_id: str) -> tuple[str
label = colored("Operation", "blue")
try:
json = flask.request.json

if json is None:
raise ValueError("No json in request")

id = json.get("operational_intent_id", "<Unknown ID>")
if json.get("operational_intent"):
op = json["operational_intent"]
Expand Down Expand Up @@ -94,6 +98,10 @@ def tracer_scd_v21_constraint_notification(observation_area_id: str) -> tuple[st
label = colored("Constraint", "magenta")
try:
json = flask.request.json

if json is None:
raise ValueError("No json in request")

id = json.get("constraint_id", "<Unknown ID>")
if json.get("constraint"):
constraint = json["constraint"]
Expand Down
Loading
Loading