371 switch from mypy to basedpyright

docs/development/adding_a_manipulator.md

@@ -9,6 +9,7 @@ read [how the system works first](../home/how_it_works.md) before proceeding.
 1. Fork the [Ephys Link repository](https://github.com/VirtualBrainLab/ephys-link).
 2. Follow the instructions for [installing Ephys Link for development](index.md#installing-for-development) to get all
    the necessary dependencies and tools set up. In this case, you'll want to clone your fork.
+3. (Optional) Familiarize yourself with the [repo's organization](code_organization.md).
 
 ## Create a Manipulator Binding
 
@@ -62,6 +63,12 @@ Use [New Scale Pathfinder MPM's binding][ephys_link.bindings.mpm_binding] as an
 Once you've implemented your binding, you can test it by running Ephys Link using your binding
 `ephys_link -b -t <cli_name>`. You can interact with it using the [Socket.IO API](socketio_api.md) or Pinpoint.
 
+## Code standards
+
+We use automatic static analyzers to check code quality. See
+the [corresponding section in the code organization documentation](code_organization.md#static-analysis) for more
+information.
+
 ## Submit Your Changes
 
 When you're satisfied with your changes, submit a pull request to the main repository. We will review your changes and

docs/development/code_organization.md

@@ -25,3 +25,17 @@ to return responses to the clients when ready.
 [`PlatformHandler`][ephys_link.back_end.platform_handler] is responsible for converting between the server API and the
 manipulator binding API. Because of this module, you don't have to worry about the details of the server API when
 writing a manipulator binding.
+
+## Static Analysis
+
+The project is strictly type checked using [`hatch fmt` (ruff)](https://hatch.pypa.io/1.9/config/static-analysis/)
+and [basedpyright](https://docs.basedpyright.com/latest/). All PRs are checked against these tools.
+
+While they are very helpful in enforcing good code, they can be annoying when working with libraries that inherently
+return `Any` (like HTTP requests) or are not strictly statically typed. In those situations we have added inline
+comments to ignore specific checks. We try to only use this in scenarios where missing typing information came from
+external sources, and it is not possible to make local type hints. Do not use file-wide ignores under any circumstances.
+We also do not make stubs since they would be challenging to maintain.
+
+We encourage using the type checker as a tool to help strengthen your code and only apply inline comments to ignore
+specific instances where external libraries cause errors. 

pyproject.toml

@@ -61,10 +61,13 @@ exclude = ["/.github", "/.idea"]
 python = "3.13.1"
 dependencies = [
     "pyinstaller==6.11.1",
+    "basedpyright==1.23.1"
 ]
 [tool.hatch.envs.default.scripts]
 exe = "pyinstaller.exe ephys_link.spec -y -- -d && pyinstaller.exe ephys_link.spec -y"
 exe-clean = "pyinstaller.exe ephys_link.spec -y --clean"
+check = "basedpyright"
+check-watched = "basedpyright --watch"
 
 [tool.hatch.envs.docs]
 python = "3.13.1"
@@ -81,7 +84,9 @@ serve = "mkdocs serve"
 build = "mkdocs build"
 
 [tool.ruff]
+exclude = ["typings"]
 unsafe-fixes = true
 
-[tool.ruff.lint]
-extend-ignore = ["DTZ005"]
+[tool.basedpyright]
+include = ["src/ephys_link"]
+strict = ["src/ephys_link"]

src/ephys_link/__main__.py

@@ -15,30 +15,31 @@
 from ephys_link.front_end.cli import CLI
 from ephys_link.front_end.gui import GUI
 from ephys_link.utils.console import Console
+from ephys_link.utils.startup import check_for_updates, preamble
 
 
 def main() -> None:
-    """Ephys Link entry point.
+    """Ephys Link entry point."""
 
-    1. Get options via CLI or GUI.
-    2. Instantiate the Console and make it globally accessible.
-    3. Instantiate the Platform Handler with the appropriate platform bindings.
-    4. Instantiate the Emergency Stop service.
-    5. Start the server.
-    """
+    # 0. Print the startup preamble.
+    preamble()
 
     # 1. Get options via CLI or GUI (if no CLI options are provided).
     options = CLI().parse_args() if len(argv) > 1 else GUI().get_options()
 
     # 2. Instantiate the Console and make it globally accessible.
     console = Console(enable_debug=options.debug)
 
-    # 3. Instantiate the Platform Handler with the appropriate platform bindings.
+    # 3. Check for updates if not disabled.
+    if not options.ignore_updates:
+        check_for_updates(console)
+
+    # 4. Instantiate the Platform Handler with the appropriate platform bindings.
     platform_handler = PlatformHandler(options, console)
 
-    # 4. Instantiate the Emergency Stop service.
+    # 5. Instantiate the Emergency Stop service.
 
-    # 5. Start the server.
+    # 6. Start the server.
     Server(options, platform_handler, console).launch()
 
 

src/ephys_link/back_end/platform_handler.py

@@ -1,4 +1,3 @@
-# ruff: noqa: BLE001
 """Manipulator platform handler.
 
 Responsible for performing the various manipulator commands.
@@ -8,6 +7,7 @@
     Instantiate PlatformHandler with the platform type and call the desired command.
 """
 
+from typing import final
 from uuid import uuid4
 
 from vbl_aquarium.models.ephys_link import (
@@ -25,11 +25,14 @@
 )
 from vbl_aquarium.models.unity import Vector4
 
+from ephys_link.bindings.mpm_binding import MPMBinding
 from ephys_link.utils.base_binding import BaseBinding
-from ephys_link.utils.common import get_bindings, vector4_to_array
 from ephys_link.utils.console import Console
+from ephys_link.utils.converters import vector4_to_array
+from ephys_link.utils.startup import get_bindings
 
 
+@final
 class PlatformHandler:
     """Handler for platform commands."""
 
@@ -73,7 +76,7 @@ def _get_binding_instance(self, options: EphysLinkOptions) -> BaseBinding:
             if binding_cli_name == options.type:
                 # Pass in HTTP port for Pathfinder MPM.
                 if binding_cli_name == "pathfinder-mpm":
-                    return binding_type(options.mpm_port)
+                    return MPMBinding(options.mpm_port)
 
                 # Otherwise just return the binding.
                 return binding_type()
@@ -103,7 +106,7 @@ async def get_platform_info(self) -> PlatformInfo:
             name=self._bindings.get_display_name(),
             cli_name=self._bindings.get_cli_name(),
             axes_count=await self._bindings.get_axes_count(),
-            dimensions=await self._bindings.get_dimensions(),
+            dimensions=self._bindings.get_dimensions(),
         )
 
     # Manipulator commands.
@@ -116,7 +119,7 @@ async def get_manipulators(self) -> GetManipulatorsResponse:
         """
         try:
             manipulators = await self._bindings.get_manipulators()
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Get Manipulators", e)
             return GetManipulatorsResponse(error=self._console.pretty_exception(e))
         else:
@@ -135,7 +138,7 @@ async def get_position(self, manipulator_id: str) -> PositionalResponse:
             unified_position = self._bindings.platform_space_to_unified_space(
                 await self._bindings.get_position(manipulator_id)
             )
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Get Position", e)
             return PositionalResponse(error=str(e))
         else:
@@ -152,7 +155,7 @@ async def get_angles(self, manipulator_id: str) -> AngularResponse:
         """
         try:
             angles = await self._bindings.get_angles(manipulator_id)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Get Angles", e)
             return AngularResponse(error=self._console.pretty_exception(e))
         else:
@@ -169,7 +172,7 @@ async def get_shank_count(self, manipulator_id: str) -> ShankCountResponse:
         """
         try:
             shank_count = await self._bindings.get_shank_count(manipulator_id)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Get Shank Count", e)
             return ShankCountResponse(error=self._console.pretty_exception(e))
         else:
@@ -214,7 +217,7 @@ async def set_position(self, request: SetPositionRequest) -> PositionalResponse:
                     )
                     self._console.error_print("Set Position", error_message)
                     return PositionalResponse(error=error_message)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Set Position", e)
             return PositionalResponse(error=self._console.pretty_exception(e))
         else:
@@ -246,7 +249,7 @@ async def set_depth(self, request: SetDepthRequest) -> SetDepthResponse:
                 )
                 self._console.error_print("Set Depth", error_message)
                 return SetDepthResponse(error=error_message)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Set Depth", e)
             return SetDepthResponse(error=self._console.pretty_exception(e))
         else:
@@ -268,7 +271,7 @@ async def set_inside_brain(self, request: SetInsideBrainRequest) -> BooleanState
                 self._inside_brain.add(request.manipulator_id)
             else:
                 self._inside_brain.discard(request.manipulator_id)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Set Inside Brain", e)
             return BooleanStateResponse(error=self._console.pretty_exception(e))
         else:
@@ -285,7 +288,7 @@ async def stop(self, manipulator_id: str) -> str:
         """
         try:
             await self._bindings.stop(manipulator_id)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Stop", e)
             return self._console.pretty_exception(e)
         else:
@@ -300,7 +303,7 @@ async def stop_all(self) -> str:
         try:
             for manipulator_id in await self._bindings.get_manipulators():
                 await self._bindings.stop(manipulator_id)
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Stop", e)
             return self._console.pretty_exception(e)
         else:

src/ephys_link/back_end/server.py

@@ -15,12 +15,12 @@
 from asyncio import get_event_loop, run
 from collections.abc import Callable, Coroutine
 from json import JSONDecodeError, dumps, loads
-from typing import Any
+from typing import Any, TypeVar, final
 from uuid import uuid4
 
 from aiohttp.web import Application, run_app
 from pydantic import ValidationError
-from socketio import AsyncClient, AsyncServer
+from socketio import AsyncClient, AsyncServer  # pyright: ignore [reportMissingTypeStubs]
 from vbl_aquarium.models.ephys_link import (
     EphysLinkOptions,
     SetDepthRequest,
@@ -32,10 +32,15 @@
 
 from ephys_link.__about__ import __version__
 from ephys_link.back_end.platform_handler import PlatformHandler
-from ephys_link.utils.common import PORT, check_for_updates, server_preamble
 from ephys_link.utils.console import Console
+from ephys_link.utils.constants import PORT
 
+# Server message generic types.
+INPUT_TYPE = TypeVar("INPUT_TYPE", bound=VBLBaseModel)
+OUTPUT_TYPE = TypeVar("OUTPUT_TYPE", bound=VBLBaseModel)
 
+
+@final
 class Server:
     def __init__(self, options: EphysLinkOptions, platform_handler: PlatformHandler, console: Console) -> None:
         """Initialize server fields based on options and platform handler.
@@ -54,12 +59,18 @@ def __init__(self, options: EphysLinkOptions, platform_handler: PlatformHandler,
         # Initialize based on proxy usage.
         self._sio: AsyncServer | AsyncClient = AsyncClient() if self._options.use_proxy else AsyncServer()
         if not self._options.use_proxy:
+            # Exit if _sio is not a Server.
+            if not isinstance(self._sio, AsyncServer):
+                error = "Server not initialized."
+                self._console.critical_print(error)
+                raise TypeError(error)
+
             self._app = Application()
-            self._sio.attach(self._app)
+            self._sio.attach(self._app)  # pyright: ignore [reportUnknownMemberType]
 
             # Bind connection events.
-            self._sio.on("connect", self.connect)
-            self._sio.on("disconnect", self.disconnect)
+            _ = self._sio.on("connect", self.connect)  # pyright: ignore [reportUnknownMemberType, reportUnknownVariableType]
+            _ = self._sio.on("disconnect", self.disconnect)  # pyright: ignore [reportUnknownMemberType, reportUnknownVariableType]
 
         # Store connected client.
         self._client_sid: str = ""
@@ -68,19 +79,13 @@ def __init__(self, options: EphysLinkOptions, platform_handler: PlatformHandler,
         self._pinpoint_id = str(uuid4())[:8]
 
         # Bind events.
-        self._sio.on("*", self.platform_event_handler)
+        _ = self._sio.on("*", self.platform_event_handler)  # pyright: ignore [reportUnknownMemberType, reportUnknownVariableType]
 
     def launch(self) -> None:
         """Launch the server.
 
         Based on the options, either connect to a proxy or launch the server locally.
         """
-        # Preamble.
-        server_preamble()
-
-        # Check for updates.
-        if not self._options.ignore_updates:
-            check_for_updates()
 
         # List platform and available manipulators.
         self._console.info_print("PLATFORM", self._platform_handler.get_display_name())
@@ -94,16 +99,22 @@ def launch(self) -> None:
             self._console.info_print("PINPOINT ID", self._pinpoint_id)
 
             async def connect_proxy() -> None:
+                # Exit if _sio is not a proxy client.
+                if not isinstance(self._sio, AsyncClient):
+                    error = "Proxy client not initialized."
+                    self._console.critical_print(error)
+                    raise TypeError(error)
+
                 # noinspection HttpUrlsUsage
-                await self._sio.connect(f"http://{self._options.proxy_address}:{PORT}")
+                await self._sio.connect(f"http://{self._options.proxy_address}:{PORT}")  # pyright: ignore [reportUnknownMemberType]
                 await self._sio.wait()
 
             run(connect_proxy())
         else:
             run_app(self._app, port=PORT)
 
     # Helper functions.
-    def _malformed_request_response(self, request: str, data: tuple[tuple[Any], ...]) -> str:
+    def _malformed_request_response(self, request: str, data: tuple[tuple[Any], ...]) -> str:  # pyright: ignore [reportExplicitAny]
         """Return a response for a malformed request.
 
         Args:
@@ -117,7 +128,10 @@ def _malformed_request_response(self, request: str, data: tuple[tuple[Any], ...]
         return dumps({"error": "Malformed request."})
 
     async def _run_if_data_available(
-        self, function: Callable[[str], Coroutine[Any, Any, VBLBaseModel]], event: str, data: tuple[tuple[Any], ...]
+        self,
+        function: Callable[[str], Coroutine[Any, Any, VBLBaseModel]],  # pyright: ignore [reportExplicitAny]
+        event: str,
+        data: tuple[tuple[Any], ...],  # pyright: ignore [reportExplicitAny]
     ) -> str:
         """Run a function if data is available.
 
@@ -136,10 +150,10 @@ async def _run_if_data_available(
 
     async def _run_if_data_parses(
         self,
-        function: Callable[[VBLBaseModel], Coroutine[Any, Any, VBLBaseModel]],
-        data_type: type[VBLBaseModel],
+        function: Callable[[INPUT_TYPE], Coroutine[Any, Any, OUTPUT_TYPE]],  # pyright: ignore [reportExplicitAny]
+        data_type: type[INPUT_TYPE],
         event: str,
-        data: tuple[tuple[Any], ...],
+        data: tuple[tuple[Any], ...],  # pyright: ignore [reportExplicitAny]
     ) -> str:
         """Run a function if data parses.
 
@@ -203,8 +217,7 @@ async def disconnect(self, sid: str) -> None:
         else:
             self._console.error_print("DISCONNECTION", f"Client {sid} disconnected without being connected.")
 
-    # noinspection PyTypeChecker
-    async def platform_event_handler(self, event: str, *args: tuple[Any]) -> str:
+    async def platform_event_handler(self, event: str, *args: tuple[Any]) -> str:  # pyright: ignore [reportExplicitAny]
         """Handle events from the server.
 
         Matches incoming events based on the Socket.IO API.