refactor(robot-server): fix/suppress type errors to enable strict mode (#8329)

Author: mcous

api/src/opentrons/hardware_control/api.py

@@ -6,6 +6,7 @@
 from collections import OrderedDict
 from typing import (
     Any,
+    Callable,
     Dict,
     Union,
     List,
@@ -45,6 +46,7 @@
     NoTipAttachedError,
     DoorState,
     DoorStateNotification,
+    HardwareEvent,
     PipettePair,
     TipAttachedError,
     HardwareAction,
@@ -289,7 +291,9 @@ def _calculate_valid_attitude(self) -> DeckTransformState:
             self._robot_calibration.deck_calibration
         )
 
-    def register_callback(self, cb):
+    def register_callback(
+        self, cb: Callable[[Union[str, HardwareEvent]], None]
+    ) -> Callable[[], None]:
         """Allows the caller to register a callback, and returns a closure
         that can be used to unregister the provided callback
         """
@@ -1883,6 +1887,6 @@ def get_instrument_max_height(
 
         return max_height
 
-    def clean_up(self):
+    def clean_up(self) -> None:
         """Get the API ready to stop cleanly."""
         self._backend.clean_up()

api/src/opentrons/hardware_control/thread_manager.py

@@ -1,13 +1,13 @@
-""" Manager for the :py:class:`.hardware_control.API` thread.
-"""
+"""Manager for the :py:class:`.hardware_control.API` thread."""
 import threading
 import logging
 import asyncio
 import functools
 import weakref
-from typing import Generic, TypeVar, Any, Optional
+from typing import Any, Awaitable, Callable, Generic, Optional, TypeVar
 from .adapters import SynchronousAdapter
 from .modules.mod_abc import AbstractModule
+from .api import API as HardwareAPI
 
 MODULE_LOG = logging.getLogger(__name__)
 
@@ -95,15 +95,20 @@ class ThreadManager:
     >>> api_single_thread.sync.home() # call as blocking sync
     """
 
-    def __init__(self, builder, *args, **kwargs):
+    def __init__(
+        self,
+        builder: Callable[..., Awaitable[HardwareAPI]],
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
         """Build the ThreadManager.
 
         :param builder: The API function to use
         """
 
-        self._loop = None
-        self.managed_obj = None
-        self.bridged_obj = None
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self.managed_obj: Optional[HardwareAPI] = None
+        self.bridged_obj: Optional[CallBridger[Any]] = None
         self._sync_managed_obj: Optional[SynchronousAdapter] = None
         is_running = threading.Event()
         self._is_running = is_running

api/src/opentrons/hardware_control/util.py

@@ -26,9 +26,9 @@ def plan_arc(
     origin_point: Point,
     dest_point: Point,
     z_height: float,
-    origin_cp: CriticalPoint = None,
-    dest_cp: CriticalPoint = None,
-    extra_waypoints: List[Tuple[float, float]] = None,
+    origin_cp: Optional[CriticalPoint] = None,
+    dest_cp: Optional[CriticalPoint] = None,
+    extra_waypoints: Optional[List[Tuple[float, float]]] = None,
 ) -> List[Tuple[Point, Optional[CriticalPoint]]]:
 
     assert z_height >= max(origin_point.z, dest_point.z)

notify-server/notify_server/models/hardware_event/__init__.py

@@ -5,3 +5,5 @@
 
 
 HardwareEventPayload = Union[DoorStatePayload]
+
+__all__ = ["HardwareEventPayload", "DoorStatePayload"]

robot-server/mypy.ini

@@ -2,43 +2,48 @@
 plugins = pydantic.mypy,decoy.mypy
 show_error_codes = True
 exclude = tests/(robot|service)
-
-# TODO(mc, 2021-09-02): replace the the following flags
-# with `strict = True` when able
-warn_unused_configs = True
-disallow_subclassing_any = True
-check_untyped_defs = True
-warn_redundant_casts = True
-strict_equality = True
-warn_unused_ignores = True
-disallow_untyped_decorators = True
-
-# TODO(mc, 2021-09-02): fix ~25 anys in generics
-# disallow_any_generics = True
-
-# TODO(mc, 2021-09-02): fix ~150 untyped call errors
-# disallow_untyped_calls = True
-
-# TODO(mc, 2021-09-02): fix ~200 untyped def errors
-# disallow_untyped_defs = True
-
-# TODO(mc, 2021-09-02): fix ~80 incomplete def errors
-# disallow_incomplete_defs = True
-
-# TODO(mc, 2021-09-02): fix ~35 implicit optionals
-# no_implicit_optional = True
-
-# TODO(mc, 2021-09-02): fix ~30 any returns
-# warn_return_any = True
-
-# TODO(mc, 2021-09-02): fix ~250 implicit re-export errors, some in FastAPI
-# FastAPI errors can be resolved by upgrading FastAPI
-# no_implicit_reexport = True
-
-# end strict mode flags
+strict = True
 
 [pydantic-mypy]
 init_forbid_extra = True
 init_typed = True
 warn_required_dynamic_aliases = True
 warn_untyped_fields = True
+
+# TODO(mc, 2021-09-08): upgrade FastAPI and remove this override
+[mypy-fastapi.*]
+no_implicit_reexport = False
+
+# TODO(mc, 2021-09-08): upgrade Pydantic and remove this override
+[mypy-pydantic.*]
+no_implicit_optional = False
+
+# TODO(mc, 2021-09-08): fix and remove any / all of the
+# overrides below whenever able
+
+[mypy-robot_server.service.*]
+# ~115 errors
+disallow_untyped_defs = False
+# ~60 errors
+disallow_untyped_calls = False
+# ~30 errors
+disallow_incomplete_defs = False
+# ~60 errors
+no_implicit_reexport = False
+
+
+[mypy-robot_server.robot.calibration.*]
+# ~115 errors
+disallow_untyped_defs = False
+# ~65 errors
+disallow_untyped_calls = False
+# ~40 errors
+disallow_incomplete_defs = False
+# ~15 errors
+warn_return_any = False
+
+# ~25 errors, total
+[mypy-tests.conftest]
+disallow_untyped_defs = False
+disallow_untyped_calls = False
+disallow_incomplete_defs = False

robot-server/robot_server/app_state.py

@@ -5,7 +5,7 @@
 """
 from fastapi import Request, WebSocket
 from starlette.datastructures import State as AppState
-from typing import Generic, Optional, TypeVar
+from typing import cast, Generic, Optional, TypeVar
 
 
 ValueT = TypeVar("ValueT")
@@ -25,7 +25,10 @@ def __init__(self, key: str) -> None:
 
     def get_from(self, app_state: AppState) -> Optional[ValueT]:
         """Get the value from state, returning None if not present."""
-        return getattr(app_state, self._key, None)
+        return cast(
+            Optional[ValueT],
+            getattr(app_state, self._key, None),
+        )
 
     def set_on(self, app_state: AppState, value: Optional[ValueT]) -> None:
         """Set the value on state."""
@@ -36,8 +39,8 @@ async def get_app_state(
     # NOTE: both of these must be typed as non-optional to allow FastAPI's
     # dependency injection magic to work, but must have default values of
     # None in order to function at runtime, as only one will be present
-    request: Request = None,
-    websocket: WebSocket = None,
+    request: Request = cast(Request, None),
+    websocket: WebSocket = cast(WebSocket, None),
 ) -> AppState:
     """Get the global application's state from the framework.
 
@@ -56,7 +59,7 @@ async def get_app_state(
         A dictionary-like object containing global application state.
     """
     request_scope = request or websocket
-    return request_scope.app.state  # type: ignore[union-attr]
+    return cast(AppState, request_scope.app.state)
 
 
 __all__ = ["AppState", "get_app_state"]

robot-server/robot_server/hardware.py

@@ -3,15 +3,15 @@
 import logging
 from pathlib import Path
 from fastapi import Depends, status
-from typing import Callable, cast
+from typing import Callable, Union, cast
 from typing_extensions import Literal
 
-from opentrons import ThreadManager, initialize as initialize_api
+from opentrons import initialize as initialize_api
 from opentrons.config import feature_flags
 from opentrons.util.helpers import utc_now
-from opentrons.hardware_control import API as HardwareAPI
+from opentrons.hardware_control import API as HardwareAPI, ThreadManager
 from opentrons.hardware_control.simulator_setup import load_simulator
-from opentrons.hardware_control.types import HardwareEvent, HardwareEventType
+from opentrons.hardware_control.types import HardwareEvent, DoorStateNotification
 
 from notify_server.clients import publisher
 from notify_server.settings import Settings as NotifyServerSettings
@@ -26,7 +26,7 @@
 log = logging.getLogger(__name__)
 
 _hw_api = AppStateValue[HardwareAPI]("hardware_api")
-_init_task = AppStateValue[asyncio.Task]("hardware_init_task")
+_init_task = AppStateValue["asyncio.Task[None]"]("hardware_init_task")
 _event_unsubscribe = AppStateValue[Callable[[], None]]("hardware_event_unsubscribe")
 
 
@@ -142,8 +142,8 @@ def _initialize_event_watchers(app_state: AppState, hardware_api: HardwareAPI) -
         notify_server_settings.publisher_address.connection_string()
     )
 
-    def _publish_hardware_event(hw_event: HardwareEvent) -> None:
-        if hw_event.event == HardwareEventType.DOOR_SWITCH_CHANGE:
+    def _publish_hardware_event(hw_event: Union[str, HardwareEvent]) -> None:
+        if isinstance(hw_event, DoorStateNotification):
             payload = DoorStatePayload(state=hw_event.new_state)
         else:
             return

robot-server/robot_server/protocols/dependencies.py

@@ -6,38 +6,38 @@
 
 from opentrons.protocol_runner import create_simulating_runner
 
-from robot_server.app_state import AppState, get_app_state
+from robot_server.app_state import AppState, AppStateValue, get_app_state
 from .protocol_store import ProtocolStore
 from .protocol_analyzer import ProtocolAnalyzer
 from .analysis_store import AnalysisStore
 
 log = logging.getLogger(__name__)
 
-_PROTOCOL_STORE_KEY = "protocol_store"
 _PROTOCOL_STORE_DIRECTORY = Path(gettempdir()) / "opentrons-protocols"
 
-_ANALYSIS_STORE_KEY = "analysis_store"
+_protocol_store = AppStateValue[ProtocolStore]("protocol_store")
+_analysis_store = AppStateValue[AnalysisStore]("analysis_store")
 
 
 def get_protocol_store(app_state: AppState = Depends(get_app_state)) -> ProtocolStore:
     """Get a singleton ProtocolStore to keep track of created protocols."""
-    protocol_store = getattr(app_state, _PROTOCOL_STORE_KEY, None)
+    protocol_store = _protocol_store.get_from(app_state)
 
     if protocol_store is None:
         log.info(f"Storing protocols in {_PROTOCOL_STORE_DIRECTORY}")
         protocol_store = ProtocolStore(directory=_PROTOCOL_STORE_DIRECTORY)
-        setattr(app_state, _PROTOCOL_STORE_KEY, protocol_store)
+        _protocol_store.set_on(app_state, protocol_store)
 
     return protocol_store
 
 
 def get_analysis_store(app_state: AppState = Depends(get_app_state)) -> AnalysisStore:
     """Get a singleton AnalysisStore to keep track of created analyses."""
-    analysis_store = getattr(app_state, _ANALYSIS_STORE_KEY, None)
+    analysis_store = _analysis_store.get_from(app_state)
 
     if analysis_store is None:
         analysis_store = AnalysisStore()
-        setattr(app_state, _ANALYSIS_STORE_KEY, analysis_store)
+        _analysis_store.set_on(app_state, analysis_store)
 
     return analysis_store
 

robot-server/robot_server/robot/calibration/check/models.py

@@ -3,6 +3,7 @@
 from functools import partial
 from pydantic import BaseModel, Field
 
+from opentrons_shared_data.labware.dev_types import LabwareDefinition
 from ..helper_classes import RequiredLabware, AttachedPipette
 
 OffsetVector = Tuple[float, float, float]
@@ -97,7 +98,7 @@ class SessionCreateParams(BaseModel):
         description="Whether to use a calibration block in the"
         "calibration health check flow.",
     )
-    tipRacks: List[dict] = Field(
+    tipRacks: List[LabwareDefinition] = Field(
         [],
         description="A list of labware definitions to use in"
         "calibration health check",
@@ -113,7 +114,7 @@ class CalibrationCheckSessionStatus(BaseModel):
     comparisonsByPipette: ComparisonStatePerPipette
     labware: List[RequiredLabware]
     activeTipRack: RequiredLabware
-    supportedCommands: List[Optional[str]] = Field(
+    supportedCommands: List[str] = Field(
         ..., description="A list of supported commands for this user flow"
     )