More mypy formatting manually

Author: khughes-bdai

spot_wrapper/cam_webrtc_client.py

@@ -64,7 +64,7 @@ def get_sdp_offer_from_spot_cam(self, token):
         result = response.json()
         return result["id"], base64.b64decode(result["sdp"]).decode()
 
-    def send_sdp_answer_to_spot_cam(self, token, offer_id, sdp_answer)-> None:
+    def send_sdp_answer_to_spot_cam(self, token, offer_id, sdp_answer) -> None:
         headers = {"Authorization": f"Bearer {token}"}
         server_url = f"https://{self.hostname}:{self.sdp_port}/{self.sdp_filename}"
 
@@ -83,26 +83,26 @@ async def start(self) -> None:
         offer_id, sdp_offer = self.get_sdp_offer_from_spot_cam(token)
 
         @self.pc.on("icegatheringstatechange")
-        def _on_ice_gathering_state_change()-> None:
+        def _on_ice_gathering_state_change() -> None:
             print(f"ICE gathering state changed to {self.pc.iceGatheringState}")
 
         @self.pc.on("signalingstatechange")
-        def _on_signaling_state_change()-> None:
+        def _on_signaling_state_change() -> None:
             print(f"Signaling state changed to: {self.pc.signalingState}")
 
         @self.pc.on("icecandidate")
-        def _on_ice_candidate(event)-> None:
+        def _on_ice_candidate(event) -> None:
             print(f"Received candidate: {event.candidate}")
 
         @self.pc.on("iceconnectionstatechange")
-        async def _on_ice_connection_state_change()-> None:
+        async def _on_ice_connection_state_change() -> None:
             print(f"ICE connection state changed to: {self.pc.iceConnectionState}")
 
             if self.pc.iceConnectionState == "checking":
                 self.send_sdp_answer_to_spot_cam(token, offer_id, self.pc.localDescription.sdp.encode())
 
         @self.pc.on("track")
-        def _on_track(track)-> None:
+        def _on_track(track) -> None:
             print(f"Received track: {track.kind}")
 
             if self.media_recorder:

spot_wrapper/cam_wrapper.py

@@ -175,7 +175,7 @@ def get_screen(self) -> str:
         """
         return self.client.get_screen()
 
-    def set_ir_colormap(self, colormap, min_temp: float, max_temp: float, auto_scale: bool=True) -> None:
+    def set_ir_colormap(self, colormap, min_temp: float, max_temp: float, auto_scale: bool = True) -> None:
         """
         Set the colormap used for the IR camera
 
@@ -188,7 +188,7 @@ def set_ir_colormap(self, colormap, min_temp: float, max_temp: float, auto_scale
         """
         self.client.set_ir_colormap(colormap, min_temp, max_temp, auto_scale)
 
-    def set_ir_meter_overlay(self, x: float, y: float, enable: bool=True) -> None:
+    def set_ir_meter_overlay(self, x: float, y: float, enable: bool = True) -> None:
         """
         Set the reticle position on the Spot CAM IR.
 
@@ -281,7 +281,7 @@ def get_volume(self) -> float:
         """
         return self.client.get_volume()
 
-    def play_sound(self, sound_name: str, gain: float=1.0) -> None:
+    def play_sound(self, sound_name: str, gain: float = 1.0) -> None:
         """
         Play a sound which is on the device
 
@@ -935,6 +935,6 @@ def __init__(self, hostname, username, password, logger, port: typing.Optional[i
 
         self._logger.info("Finished setting up spot cam wrapper components")
 
-    def shutdown(self)-> None:
+    def shutdown(self) -> None:
         self._logger.info("Shutting down Spot CAM wrapper")
         self.image.shutdown_flag.set()

spot_wrapper/spot_arm.py

@@ -136,7 +136,7 @@ def ensure_arm_power_and_stand(self) -> typing.Tuple[bool, str]:
 
         return True, "Spot has an arm, is powered on, and standing"
 
-    def wait_for_arm_command_to_complete(self, cmd_id, timeout_sec: typing.Optional[float]=None) -> None:
+    def wait_for_arm_command_to_complete(self, cmd_id, timeout_sec: typing.Optional[float] = None) -> None:
         """
         Wait until a command issued to the arm complets. Wrapper around the SDK function for convenience
 
@@ -295,7 +295,9 @@ def arm_joint_move(self, joint_targets: typing.List[float]) -> typing.Tuple[bool
         except Exception as e:
             return False, f"Exception occured during arm movement: {e}"
 
-    def create_wrench_from_forces_and_torques(self, forces: typing.List[float], torques: typing.List[float]) -> geometry_pb2.Wrench:
+    def create_wrench_from_forces_and_torques(
+        self, forces: typing.List[float], torques: typing.List[float]
+    ) -> geometry_pb2.Wrench:
         force = geometry_pb2.Vec3(x=forces[0], y=forces[1], z=forces[2])
         torque = geometry_pb2.Vec3(x=torques[0], y=torques[1], z=torques[2])
         return geometry_pb2.Wrench(force=force, torque=torque)

spot_wrapper/spot_eap.py

@@ -47,6 +47,7 @@ def _start_query(self) -> typing.Optional[FutureWrapper]:
             callback_future = self._client.get_point_cloud_async(self._point_cloud_requests)
             callback_future.add_done_callback(self._callback)
             return callback_future
+        return None
 
 
 class SpotEAP:

spot_wrapper/spot_graph_nav.py

@@ -51,7 +51,7 @@ def _get_lease(self) -> Lease:
         self._lease = self._lease_wallet.get_lease()
         return self._lease
 
-    def _init_current_graph_nav_state(self):
+    def _init_current_graph_nav_state(self) -> None:
         # Store the most recent knowledge of the state of the robot based on rpc calls.
         self._current_graph = None
         self._current_edges = {}  # maps to_waypoint to list(from_waypoint)
@@ -75,7 +75,7 @@ def navigate_initial_localization(
         upload_path: str,
         initial_localization_fiducial: bool = True,
         initial_localization_waypoint: typing.Optional[str] = None,
-    ):
+    ) -> typing.Tuple[bool, str]:
         """Navigate with graphnav.
 
         Args:
@@ -112,7 +112,7 @@ def navigate_initial_localization(
 
         return True, "Localization done!"
 
-    def navigate_to_existing_waypoint(self, waypoint_id: str):
+    def navigate_to_existing_waypoint(self, waypoint_id: str) -> typing.Tuple[bool, str]:
         """Navigate to an existing waypoint.
         Args:
             waypoint_id : Waypoint id string for where to go
@@ -121,7 +121,7 @@ def navigate_to_existing_waypoint(self, waypoint_id: str):
         resp = self._navigate_to(waypoint_id)
         return resp
 
-    def navigate_through_route(self, waypoint_ids: typing.List[str]):
+    def navigate_through_route(self, waypoint_ids: typing.List[str]) -> typing.Tuple[bool, str]:
         """
         Args:
             waypoint_ids: List[str] of waypoints to follow
@@ -146,7 +146,7 @@ def navigation_close_loops(self, close_fiducial_loops: bool, close_odometry_loop
     def optmize_anchoring(self) -> typing.Tuple[bool, str]:
         return self._optimize_anchoring()
 
-    def _write_bytes_while_download(self, filepath: str, data: bytes):
+    def _write_bytes_while_download(self, filepath: str, data: bytes) -> None:
         """Write data to a file.
         Args:
             filepath (str) : Path of file where data will be written.
@@ -257,14 +257,14 @@ def download_graph(self, download_path: str) -> typing.Tuple[bool, str]:
     # Downloading, reproducing, distributing or otherwise using the SDK Software
     # is subject to the terms and conditions of the Boston Dynamics Software
     # Development Kit License (20191101-BDSDK-SL).
-    def _get_localization_state(self, *args):
+    def _get_localization_state(self, *args) -> None:
         """Get the current localization and state of the robot."""
         state = self._graph_nav_client.get_localization_state()
         self._logger.info(f"Got localization: \n{str(state.localization)}")
         odom_tform_body = get_odom_tform_body(state.robot_kinematics.transforms_snapshot)
         self._logger.info(f"Got robot state in kinematic odometry frame: \n{str(odom_tform_body)}")
 
-    def set_initial_localization_fiducial(self, *args):
+    def set_initial_localization_fiducial(self, *args) -> None:
         """Trigger localization when near a fiducial."""
         robot_state = self._robot_state_client.get_robot_state()
         current_odom_tform_body = get_odom_tform_body(robot_state.kinematic_state.transforms_snapshot).to_proto()
@@ -276,7 +276,7 @@ def set_initial_localization_fiducial(self, *args):
             ko_tform_body=current_odom_tform_body,
         )
 
-    def set_initial_localization_waypoint(self, *args):
+    def set_initial_localization_waypoint(self, *args) -> None:
         """Trigger localization to a waypoint."""
         # Take the first argument as the localization waypoint.
         if len(args) < 1:
@@ -316,7 +316,7 @@ def _download_current_graph(self):
         self._current_graph = graph
         return graph
 
-    def _download_full_graph(self, *args):
+    def _download_full_graph(self, *args) -> None:
         """Download the graph and snapshots from the robot."""
         graph = self._graph_nav_client.download_graph()
         if graph is None:
@@ -330,12 +330,12 @@ def _download_full_graph(self, *args):
         self._download_and_write_waypoint_snapshots(graph.waypoints)
         self._download_and_write_edge_snapshots(graph.edges)
 
-    def _write_full_graph(self, graph):
+    def _write_full_graph(self, graph) -> None:
         """Download the graph from robot to the specified, local filepath location."""
         graph_bytes = graph.SerializeToString()
         self._write_bytes(self._download_filepath, "/graph", graph_bytes)
 
-    def _download_and_write_waypoint_snapshots(self, waypoints):
+    def _download_and_write_waypoint_snapshots(self, waypoints) -> None:
         """Download the waypoint snapshots from robot to the specified, local filepath location."""
         num_waypoint_snapshots_downloaded = 0
         for waypoint in waypoints:
@@ -359,7 +359,7 @@ def _download_and_write_waypoint_snapshots(self, waypoints):
                 )
             )
 
-    def _download_and_write_edge_snapshots(self, edges):
+    def _download_and_write_edge_snapshots(self, edges) -> None:
         """Download the edge snapshots from robot to the specified, local filepath location."""
         num_edge_snapshots_downloaded = 0
         num_to_download = 0
@@ -383,7 +383,7 @@ def _download_and_write_edge_snapshots(self, edges):
                 "Downloaded {} of the total {} edge snapshots.".format(num_edge_snapshots_downloaded, num_to_download)
             )
 
-    def _write_bytes(self, filepath: str, filename: str, data):
+    def _write_bytes(self, filepath: str, filename: str, data) -> None:
         """Write data to a file."""
         os.makedirs(filepath, exist_ok=True)
         with open(filepath + filename, "wb+") as f:
@@ -655,7 +655,9 @@ def _match_edge(
                     return map_pb2.Edge.Id(from_waypoint=waypoint1, to_waypoint=waypoint2)
         return None
 
-    def _auto_close_loops(self, close_fiducial_loops: bool, close_odometry_loops: bool, *args):
+    def _auto_close_loops(
+        self, close_fiducial_loops: bool, close_odometry_loops: bool, *args
+    ) -> typing.Tuple[bool, str]:
         """Automatically find and close all loops in the graph."""
         response: map_processing_pb2.ProcessTopologyResponse = self._map_processing_client.process_topology(
             params=map_processing_pb2.ProcessTopologyRequest.Params(
@@ -676,7 +678,7 @@ def _auto_close_loops(self, close_fiducial_loops: bool, close_odometry_loops: bo
         else:
             return False, "Unknown error during map processing."
 
-    def _optimize_anchoring(self, *args):
+    def _optimize_anchoring(self, *args) -> typing.Tuple[bool, str]:
         """Call anchoring optimization on the server, producing a globally optimal reference frame for waypoints to be
         expressed in.
         """
@@ -712,7 +714,7 @@ def _optimize_anchoring(self, *args):
                 return False, status_mapping[response.status]
             return False, "Unknown error during anchoring optimization."
 
-    def _id_to_short_code(self, id: str):
+    def _id_to_short_code(self, id: str) -> typing.Optional[str]:
         """Convert a unique id to a 2 letter short code."""
         tokens = id.split("-")
         if len(tokens) > 2:
@@ -726,7 +728,7 @@ def _pretty_print_waypoints(
         short_code_to_count: typing.Dict[str, int],
         localization_id: str,
         logger: logging.Logger,
-    ):
+    ) -> None:
         short_code = self._id_to_short_code(waypoint_id)
         if short_code is None or short_code_to_count[short_code] != 1:
             # If the short code is not valid/unique, don't show it.

spot_wrapper/spot_images.py

@@ -216,7 +216,7 @@ def __init__(
                 continue
             self._image_requests_by_camera[camera] = {}
             image_types = IMAGE_SOURCES_BY_CAMERA[camera]
-            quality = 75
+            quality = 75.0
             for image_type in image_types:
                 if image_type.startswith("depth"):
                     image_format = image_pb2.Image.FORMAT_RAW

spot_wrapper/tests/test_graph_nav_util.py

@@ -16,13 +16,13 @@ def __init__(self) -> None:
 
 
 class TestGraphNavUtilShortCode:
-    def test_id_to_short_code(self):
+    def test_id_to_short_code(self) -> None:
         assert graph_nav_util._id_to_short_code("ebony-pug-mUzxLNq.TkGlVIxga+UKAQ==") == "ep"
         assert graph_nav_util._id_to_short_code("erose-simian-sug9xpxhCxgft7Mtbhr98A==") == "es"
 
 
 class TestGraphNavUtilFindUniqueWaypointId:
-    def test_short_code(self):
+    def test_short_code(self) -> None:
         # Set up
         self.logger = logging.Logger("test_graph_nav_util", level=logging.INFO)
         self.graph = map_pb2.Graph()
@@ -34,7 +34,7 @@ def test_short_code(self):
         # Test annotation name that is unknown
         assert graph_nav_util._find_unique_waypoint_id("ABCDEF", self.graph, self.name_to_id, self.logger) == "ABCDEF"
 
-    def test_short_code_with_graph(self):
+    def test_short_code_with_graph(self) -> None:
         # Set up
         self.logger = logging.Logger("test_graph_nav_util", level=logging.INFO)
         self.graph = map_pb2.Graph()
@@ -51,7 +51,7 @@ def test_short_code_with_graph(self):
 
 
 class TestGraphNavUtilUpdateWaypointsEdges:
-    def test_empty_graph(self):
+    def test_empty_graph(self) -> None:
         self.logger = logging.Logger("test_graph_nav_util", level=logging.INFO)
 
         # Test empty graph
@@ -61,7 +61,7 @@ def test_empty_graph(self):
         assert len(self.graph.waypoints) == 0
         assert len(self.graph.edges) == 0
 
-    def test_one_waypoint(self):
+    def test_one_waypoint(self) -> None:
         self.logger = logging.Logger("test_graph_nav_util", level=logging.INFO)
 
         # Test graph with 1 waypoint
@@ -80,7 +80,7 @@ def test_one_waypoint(self):
         assert len(self.name_to_id) == 1
         assert self.name_to_id["Node1"] == "ABCDE"
 
-    def test_two_waypoints_with_edge(self):
+    def test_two_waypoints_with_edge(self) -> None:
         self.logger = logging.Logger("test_graph_nav_util", level=logging.INFO)
 
         # Test graph with 2 waypoints and an edge between them
@@ -107,7 +107,7 @@ def test_two_waypoints_with_edge(self):
         assert self.name_to_id["Node1"] == "ABCDE"
         assert self.name_to_id["Node2"] == "DE"
 
-    def test_two_waypoints_with_edge_and_localization(self):
+    def test_two_waypoints_with_edge_and_localization(self) -> None:
         self.logger = logging.Logger("test_graph_nav_util", level=logging.INFO)
 
         # Test graph with 2 waypoints and an edge between them. Mainly affects the pretty print.

spot_wrapper/wrapper.py

@@ -101,7 +101,7 @@ def robotToLocalTime(timestamp: Timestamp, robot: Robot) -> Timestamp:
 class MissingSpotArm(Exception):
     """Raised when the arm is not available on the robot"""
 
-    def __init__(self, message: str ="Spot arm not available"):
+    def __init__(self, message: str = "Spot arm not available"):
         # Call the base class constructor with the parameters it needs
         super().__init__(message)