Add typehinting to functions missing this

src/isar/config/keyvault/keyvault_service.py

@@ -1,5 +1,6 @@
 import logging
 import traceback
+from typing import Union
 
 from azure.core.exceptions import (
     ClientAuthenticationError,
@@ -52,16 +53,17 @@ def set_secret(self, secret_name: str, secret_value) -> None:
             traceback.print_exc()
             raise KeyvaultError  # type: ignore
 
-    def get_secret_client(self):
+    def get_secret_client(self) -> SecretClient:
         try:
+            credential: Union[ClientSecretCredential, DefaultAzureCredential]
             if self.client_id and self.client_secret and self.tenant_id:
-                credential: ClientSecretCredential = ClientSecretCredential(
+                credential = ClientSecretCredential(
                     tenant_id=self.tenant_id,
                     client_id=self.client_id,
                     client_secret=self.client_secret,
                 )
             else:
-                credential: DefaultAzureCredential = DefaultAzureCredential()
+                credential = DefaultAzureCredential()
         except ClientAuthenticationError:
             self.logger.error("Failed to authenticate to Azure.")
             traceback.print_exc()

src/isar/config/log.py

@@ -8,7 +8,7 @@
 from isar.config.settings import settings
 
 
-def setup_logger():
+def setup_logger() -> None:
     log_levels: dict = settings.LOG_LEVELS
     with pkg_resources.path("isar.config", "logging.conf") as path:
         log_config = yaml.safe_load(open(path))

src/isar/config/settings.py

@@ -307,7 +307,7 @@ def __init__(self) -> None:
 
     # Model of the robot which ISAR is connected to
     # This should be set in the robot package settings.env file
-    ROBOT_MODEL: RobotModel = Field(default=RobotModel.Robot)
+    ROBOT_MODEL: RobotModel = Field(default=RobotModel.Robot)  # type: ignore
 
     class Config:
         env_file_encoding = "utf-8"

src/isar/models/communication/queues/queues.py

@@ -6,7 +6,7 @@
 
 
 class Queues:
-    def __init__(self):
+    def __init__(self) -> None:
         self.start_mission: QueueIO = QueueIO(input_size=1, output_size=1)
         self.stop_mission: QueueIO = QueueIO(input_size=1, output_size=1)
         self.pause_mission: QueueIO = QueueIO(input_size=1, output_size=1)

src/isar/services/service_connections/mqtt/robot_info_publisher.py

@@ -17,7 +17,7 @@ def run(self) -> None:
         while True:
             payload: RobotInfoPayload = RobotInfoPayload(
                 robot_name=settings.ROBOT_ID,
-                robot_model=robot_settings.ROBOT_MODEL,
+                robot_model=robot_settings.ROBOT_MODEL,  # type: ignore
                 robot_serial_number=settings.SERIAL_NUMBER,
                 video_streams=settings.VIDEO_STREAMS,
                 host=settings.API_HOST_VIEWED_EXTERNALLY,

src/isar/services/utilities/threaded_request.py

@@ -3,7 +3,7 @@
 
 
 class ThreadedRequest:
-    def __init__(self, request_func: Any) -> None:
+    def __init__(self, request_func: Any):
         self._thread: Optional[Thread] = None
         self._request_func: Any = request_func
         self._output: Optional[Any] = None

src/isar/state_machine/states/idle.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import TYPE_CHECKING, Optional
+from typing import Optional, TYPE_CHECKING
 
 from transitions import State
 
@@ -11,19 +11,19 @@
 
 
 class Idle(State):
-    def __init__(self, state_machine: "StateMachine"):
+    def __init__(self, state_machine: "StateMachine") -> None:
         super().__init__(name="idle", on_enter=self.start, on_exit=self.stop)
         self.state_machine: "StateMachine" = state_machine
         self.logger = logging.getLogger("state_machine")
 
-    def start(self):
+    def start(self) -> None:
         self.state_machine.update_state()
         self._run()
 
-    def stop(self):
+    def stop(self) -> None:
         pass
 
-    def _run(self):
+    def _run(self) -> None:
         while True:
             start_mission: Optional[
                 StartMissionMessage
@@ -33,7 +33,7 @@ def _run(self):
                     mission=start_mission.mission,
                     initial_pose=start_mission.initial_pose,
                 )
-                transition = self.state_machine.mission_started
+                transition = self.state_machine.mission_started  # type: ignore
                 break
             time.sleep(self.state_machine.sleep_time)
 

src/isar/state_machine/states/initialize.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import TYPE_CHECKING, Callable, Optional
+from typing import Callable, Optional, TYPE_CHECKING
 
 from injector import inject
 from transitions import State
@@ -17,23 +17,23 @@
 
 class Initialize(State):
     @inject
-    def __init__(self, state_machine: "StateMachine"):
+    def __init__(self, state_machine: "StateMachine") -> None:
         super().__init__(name="initialize", on_enter=self.start, on_exit=self.stop)
         self.state_machine: "StateMachine" = state_machine
 
         self.logger = logging.getLogger("state_machine")
         self.initialize_thread: Optional[ThreadedRequest] = None
 
-    def start(self):
+    def start(self) -> None:
         self.state_machine.update_state()
         self._run()
 
-    def stop(self):
+    def stop(self) -> None:
         if self.initialize_thread:
             self.initialize_thread.wait_for_thread()
         self.initialize_thread = None
 
-    def _run(self):
+    def _run(self) -> None:
         transition: Callable
         while True:
             if not self.initialize_thread:
@@ -52,9 +52,9 @@ def _run(self):
                 continue
             except RobotException as e:
                 self.logger.error(f"Initialization of robot failed. Error: {e}")
-                transition = self.state_machine.initialization_failed
+                transition = self.state_machine.initialization_failed  # type: ignore
                 break
 
-            transition = self.state_machine.initialization_successful
+            transition = self.state_machine.initialization_successful  # type: ignore
             break
         transition()

src/isar/state_machine/states/initiate_step.py

@@ -1,28 +1,25 @@
 import logging
 import time
-from typing import TYPE_CHECKING, Callable
+from typing import Callable, Optional, TYPE_CHECKING
 
 from transitions import State
 
 from isar.config.settings import settings
-from isar.models.mission.status import MissionStatus
 from isar.services.utilities.threaded_request import (
     ThreadedRequest,
     ThreadedRequestNotFinishedError,
 )
-from isar.state_machine.states_enum import States
 from robot_interface.models.exceptions import (
     RobotException,
     RobotInfeasibleStepException,
 )
-from robot_interface.models.mission.status import StepStatus
 
 if TYPE_CHECKING:
     from isar.state_machine.state_machine import StateMachine
 
 
 class InitiateStep(State):
-    def __init__(self, state_machine: "StateMachine"):
+    def __init__(self, state_machine: "StateMachine") -> None:
         super().__init__(name="initiate_step", on_enter=self.start, on_exit=self.stop)
         self.state_machine: "StateMachine" = state_machine
         self.initiate_step_failure_counter: int = 0
@@ -31,34 +28,34 @@ def __init__(self, state_machine: "StateMachine"):
         )
         self.logger = logging.getLogger("state_machine")
 
-        self.initiate_step_thread = None
+        self.initiate_step_thread: Optional[ThreadedRequest] = None
 
-    def start(self):
+    def start(self) -> None:
         self.state_machine.update_state()
         self._run()
 
-    def stop(self):
+    def stop(self) -> None:
         self.initiate_step_failure_counter = 0
         if self.initiate_step_thread:
             self.initiate_step_thread.wait_for_thread()
         self.initiate_step_thread = None
 
-    def _run(self):
+    def _run(self) -> None:
         transition: Callable
         while True:
             if self.state_machine.should_stop_mission():
-                transition = self.state_machine.stop
+                transition = self.state_machine.stop  # type: ignore
                 break
 
             if self.state_machine.should_pause_mission():
-                transition = self.state_machine.pause
+                transition = self.state_machine.pause  # type: ignore
                 break
 
             if not self.state_machine.current_task:
                 self.logger.info(
                     f"Completed mission: {self.state_machine.current_mission.id}"
                 )
-                transition = self.state_machine.mission_finished
+                transition = self.state_machine.mission_finished  # type: ignore
                 break
 
             if not self.initiate_step_thread:
@@ -72,7 +69,7 @@ def _run(self):
 
             try:
                 self.initiate_step_thread.get_output()
-                transition = self.state_machine.step_initiated
+                transition = self.state_machine.step_initiated  # type: ignore
                 break
             except ThreadedRequestNotFinishedError:
                 time.sleep(self.state_machine.sleep_time)
@@ -83,7 +80,7 @@ def _run(self):
                     f"{type(self.state_machine.current_step).__name__}"
                     f"Invalid step: {str(self.state_machine.current_step.id)[:8]}"
                 )
-                transition = self.state_machine.step_infeasible
+                transition = self.state_machine.step_infeasible  # type: ignore
                 break
             except RobotException as e:
                 self.initiate_step_thread = None
@@ -103,7 +100,7 @@ def _run(self):
                     f"{self.initiate_step_failure_counter_limit} attempts. "
                     f"Cancelling mission."
                 )
-                transition = self.state_machine.initiate_step_failed
+                transition = self.state_machine.initiate_step_failed  # type: ignore
                 break
 
             time.sleep(self.state_machine.sleep_time)

src/isar/state_machine/states/monitor.py

@@ -1,7 +1,7 @@
 import logging
 import time
 from copy import deepcopy
-from typing import TYPE_CHECKING, Callable, Optional, Sequence, Tuple
+from typing import Callable, Optional, Sequence, TYPE_CHECKING, Tuple
 
 from injector import inject
 from transitions import State
@@ -21,31 +21,31 @@
 
 class Monitor(State):
     @inject
-    def __init__(self, state_machine: "StateMachine"):
+    def __init__(self, state_machine: "StateMachine") -> None:
         super().__init__(name="monitor", on_enter=self.start, on_exit=self.stop)
         self.state_machine: "StateMachine" = state_machine
 
         self.logger = logging.getLogger("state_machine")
         self.step_status_thread: Optional[ThreadedRequest] = None
 
-    def start(self):
+    def start(self) -> None:
         self.state_machine.update_state()
         self._run()
 
-    def stop(self):
+    def stop(self) -> None:
         if self.step_status_thread:
             self.step_status_thread.wait_for_thread()
         self.step_status_thread = None
 
-    def _run(self):
+    def _run(self) -> None:
         transition: Callable
         while True:
             if self.state_machine.should_stop_mission():
-                transition = self.state_machine.stop
+                transition = self.state_machine.stop  # type: ignore
                 break
 
             if self.state_machine.should_pause_mission():
-                transition = self.state_machine.pause
+                transition = self.state_machine.pause  # type: ignore
                 break
 
             if not self.step_status_thread:
@@ -72,15 +72,15 @@ def _run(self):
                     self.state_machine.current_step,
                     name="State Machine Get Inspections",
                 )
-                transition = self.state_machine.step_finished
+                transition = self.state_machine.step_finished  # type: ignore
                 break
 
             self.step_status_thread = None
             time.sleep(self.state_machine.sleep_time)
 
         transition()
 
-    def _queue_inspections_for_upload(self, current_step: InspectionStep):
+    def _queue_inspections_for_upload(self, current_step: InspectionStep) -> None:
         try:
             inspections: Sequence[
                 Inspection

src/isar/state_machine/states/paused.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import TYPE_CHECKING, Callable
+from typing import Callable, TYPE_CHECKING
 
 from transitions import State
 
@@ -9,24 +9,24 @@
 
 
 class Paused(State):
-    def __init__(self, state_machine: "StateMachine"):
+    def __init__(self, state_machine: "StateMachine") -> None:
         super().__init__(name="paused", on_enter=self.start)
         self.state_machine: "StateMachine" = state_machine
         self.logger = logging.getLogger("state_machine")
 
-    def start(self):
+    def start(self) -> None:
         self.state_machine.update_state()
         self._run()
 
-    def _run(self):
+    def _run(self) -> None:
         transition: Callable
         while True:
             if self.state_machine.should_stop_mission():
-                transition = self.state_machine.mission_stopped
+                transition = self.state_machine.mission_stopped  # type: ignore
                 break
 
             if self.state_machine.should_resume_mission():
-                transition = self.state_machine.resume
+                transition = self.state_machine.resume  # type: ignore
                 break
 
             time.sleep(self.state_machine.sleep_time)