Remove Optional and Union type hint

Author: andchiind

docs/update_state_diagram.py

@@ -1,11 +1,11 @@
 import ast
 import os
-from typing import List, Optional
+from typing import List
 
 from python_to_mermaid import MermaidDiagram
 
 
-def get_imports(source_code: str) -> tuple[Optional[str], list]:
+def get_imports(source_code: str) -> tuple[str | None, list]:
     own_class_name = None
     imported_states = []
     tree = ast.parse(source_code)

src/isar/apis/api.py

@@ -3,7 +3,7 @@
 from datetime import datetime, timezone
 from http import HTTPStatus
 from logging import Logger
-from typing import Any, List, Union
+from typing import Any, List
 
 import click
 import uvicorn
@@ -82,7 +82,7 @@ def _create_app(self) -> FastAPI:
         )
 
         if self.authenticator.should_authenticate():
-            backend_cors_origins: List[Union[str, AnyHttpUrl]] = [
+            backend_cors_origins: List[str | AnyHttpUrl] = [
                 f"http://{self.host}:{self.port}"
             ]
 

src/isar/apis/models/models.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import List
 
 from alitra import Frame, Orientation, Pose, Position
 from pydantic import BaseModel, Field
@@ -8,8 +8,8 @@
 
 class TaskResponse(BaseModel):
     id: str
-    tag_id: Optional[str] = None
-    inspection_id: Optional[str] = None
+    tag_id: str | None = None
+    inspection_id: str | None = None
     type: TaskTypes
 
 
@@ -20,23 +20,23 @@ class StartMissionResponse(BaseModel):
 
 class ControlMissionResponse(BaseModel):
     success: bool
-    failure_reason: Optional[str] = None
+    failure_reason: str | None = None
 
 
 class MissionStartResponse(BaseModel):
-    mission_id: Optional[str] = None
+    mission_id: str | None = None
     mission_started: bool
-    mission_not_started_reason: Optional[str] = None
+    mission_not_started_reason: str | None = None
 
 
 class LockdownResponse(BaseModel):
     lockdown_started: bool
-    failure_reason: Optional[str] = None
+    failure_reason: str | None = None
 
 
 class MaintenanceResponse(BaseModel):
     is_maintenance_mode: bool
-    failure_reason: Optional[str] = None
+    failure_reason: str | None = None
 
 
 class RobotInfoResponse(BaseModel):

src/isar/apis/models/start_mission_definition.py

@@ -1,6 +1,6 @@
 import time
 from enum import Enum
-from typing import List, Optional
+from typing import List
 from uuid import uuid4
 
 from pydantic import BaseModel, Field
@@ -38,28 +38,28 @@ class TaskType(str, Enum):
 class StartMissionInspectionDefinition(BaseModel):
     type: InspectionTypes = Field(default=InspectionTypes.image)
     inspection_target: InputPosition
-    inspection_description: Optional[str] = None
-    duration: Optional[float] = None
+    inspection_description: str | None = None
+    duration: float | None = None
 
 
 class StartMissionTaskDefinition(BaseModel):
-    id: Optional[str] = None
+    id: str | None = None
     type: TaskType = Field(default=TaskType.Inspection)
     pose: InputPose
-    inspection: Optional[StartMissionInspectionDefinition] = None
-    tag: Optional[str] = None
-    zoom: Optional[ZoomDescription] = None
+    inspection: StartMissionInspectionDefinition | None = None
+    tag: str | None = None
+    zoom: ZoomDescription | None = None
 
 
 class StartMissionDefinition(BaseModel):
-    id: Optional[str] = None
+    id: str | None = None
     tasks: List[StartMissionTaskDefinition]
-    name: Optional[str] = None
-    start_pose: Optional[InputPose] = None
+    name: str | None = None
+    start_pose: InputPose | None = None
 
 
 class StopMissionDefinition(BaseModel):
-    mission_id: Optional[str] = None
+    mission_id: str | None = None
 
 
 class MissionFormatError(Exception):

src/isar/apis/robot_control/robot_controller.py

@@ -1,5 +1,4 @@
 import logging
-from typing import Optional
 
 from fastapi import HTTPException
 from opentelemetry import trace
@@ -22,9 +21,7 @@ def __init__(
 
     @tracer.start_as_current_span("generate_media_config")
     def generate_media_config(self) -> MediaConfig:
-        media_config: Optional[MediaConfig] = (
-            self.robot_utilities.generate_media_config()
-        )
+        media_config: MediaConfig | None = self.robot_utilities.generate_media_config()
         if media_config is None:
             raise HTTPException(
                 status_code=204,

src/isar/apis/security/authentication.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Callable, Coroutine, Type, Union
+from typing import Any, Callable, Coroutine, Type
 
 from fastapi import Depends
 from fastapi.security.base import SecurityBase
@@ -56,7 +56,7 @@ def should_authenticate(self) -> bool:
 
     def get_scheme(
         self,
-    ) -> Union[Callable[[Any], Coroutine[Any, Any, None]], Type[NoSecurity]]:
+    ) -> Callable[[Any], Coroutine[Any, Any, None]] | Type[NoSecurity]:
         if self.should_authenticate():
             return validate_has_role
         return NoSecurity

src/isar/eventhandlers/eventhandler.py

@@ -4,7 +4,7 @@
 from copy import deepcopy
 from dataclasses import dataclass
 from threading import Event as ThreadEvent
-from typing import TYPE_CHECKING, Any, Callable, Generic, List, Optional, TypeVar
+from typing import TYPE_CHECKING, Any, Callable, Generic, List, TypeVar
 
 from isar.config.settings import settings
 from isar.models.events import Event
@@ -20,15 +20,15 @@
 class EventHandlerMapping(Generic[T]):
     name: str
     event: Event[T]
-    handler: Callable[[T], Optional[Transition]]
+    handler: Callable[[T], Transition | None]
     should_not_consume: bool = False
 
 
 @dataclass
 class TimeoutHandlerMapping:
     name: str
     timeout_in_seconds: float
-    handler: Callable[[], Optional[Transition]]
+    handler: Callable[[], Transition | None]
 
 
 if TYPE_CHECKING:
@@ -55,7 +55,7 @@ def __init__(
 
     def get_event_handler_by_name(
         self, event_handler_name: str
-    ) -> Optional[EventHandlerMapping]:
+    ) -> EventHandlerMapping | None:
         filtered_handlers = list(
             filter(
                 lambda mapping: mapping.name == event_handler_name,
@@ -66,7 +66,7 @@ def get_event_handler_by_name(
 
     def get_event_timer_by_name(
         self, event_timer_name: str
-    ) -> Optional[TimeoutHandlerMapping]:
+    ) -> TimeoutHandlerMapping | None:
         filtered_timers = list(
             filter(
                 lambda mapping: mapping.name == event_timer_name,
@@ -75,7 +75,7 @@ def get_event_timer_by_name(
         )
         return filtered_timers[0] if len(filtered_timers) > 0 else None
 
-    def run(self) -> Optional["State"]:
+    def run(self) -> "State | None":
         should_exit_state: bool = False
         timers = deepcopy(self.timers)
         entered_time = time.time()
@@ -95,7 +95,7 @@ def run(self) -> Optional["State"]:
                 break
 
             for handler_mapping in self.event_handler_mappings:
-                event_value: Optional[Any]
+                event_value: Any | None
                 if handler_mapping.should_not_consume:
                     event_value = handler_mapping.event.check()
                 else:

src/isar/models/events.py

@@ -1,7 +1,7 @@
 from collections import deque
 from queue import Empty, Queue
 from threading import Lock
-from typing import Generic, Optional, Tuple, TypeVar
+from typing import Generic, Tuple, TypeVar
 
 from isar.apis.models.models import (
     ControlMissionResponse,
@@ -35,7 +35,7 @@ def __init__(self, name: str) -> None:
         super().__init__(maxsize=1)
         self.name = name
 
-    def trigger_event(self, data: T, timeout: Optional[int] = None) -> None:
+    def trigger_event(self, data: T, timeout: int | None = None) -> None:
         try:
             # We always want a timeout when blocking for results, so that
             # the thread will never get stuck waiting for a result
@@ -45,7 +45,7 @@ def trigger_event(self, data: T, timeout: Optional[int] = None) -> None:
                 raise EventTimeoutError
             return None
 
-    def consume_event(self, timeout: Optional[int] = None) -> Optional[T]:
+    def consume_event(self, timeout: int | None = None) -> T | None:
         try:
             return self.get(block=timeout is not None, timeout=timeout)
         except Empty:
@@ -69,7 +69,7 @@ def has_event(self) -> bool:
             self.qsize() != 0
         )  # Queue size is not reliable, but should be sufficient for this case
 
-    def check(self) -> Optional[T]:
+    def check(self) -> T | None:
         if not self._qsize():
             return None
         with self.mutex:

src/isar/robot/robot.py

@@ -2,7 +2,7 @@
 from queue import Queue
 from threading import Event as ThreadEvent
 from threading import Thread
-from typing import Callable, List, Optional, Tuple
+from typing import Callable, List, Tuple
 
 from isar.config.settings import settings
 from isar.models.events import (
@@ -45,17 +45,17 @@ def __init__(
         self.upload_queue: Queue = events.upload_queue
         self.shared_state: SharedState = shared_state
         self.robot: RobotInterface = robot
-        self.start_mission_thread: Optional[RobotStartMissionThread] = None
-        self.robot_battery_thread: Optional[RobotBatteryThread] = None
-        self.robot_status_thread: Optional[RobotStatusThread] = None
-        self.monitor_mission_thread: Optional[RobotMonitorMissionThread] = None
-        self.stop_mission_thread: Optional[RobotStopMissionThread] = None
-        self.pause_mission_thread: Optional[RobotPauseMissionThread] = None
-        self.resume_mission_thread: Optional[RobotResumeMissionThread] = None
+        self.start_mission_thread: RobotStartMissionThread | None = None
+        self.robot_battery_thread: RobotBatteryThread | None = None
+        self.robot_status_thread: RobotStatusThread | None = None
+        self.monitor_mission_thread: RobotMonitorMissionThread | None = None
+        self.stop_mission_thread: RobotStopMissionThread | None = None
+        self.pause_mission_thread: RobotPauseMissionThread | None = None
+        self.resume_mission_thread: RobotResumeMissionThread | None = None
         self.upload_inspection_threads: List[RobotUploadInspectionThread] = []
         self.signal_thread_quitting: ThreadEvent = ThreadEvent()
         self.signal_mission_stopped: ThreadEvent = ThreadEvent()
-        self.inspection_callback_thread: Optional[Thread] = None
+        self.inspection_callback_thread: Thread | None = None
 
     def stop(self) -> None:
         self.signal_thread_quitting.set()

src/isar/robot/robot_monitor_mission.py

@@ -1,7 +1,7 @@
 import logging
 import time
 from threading import Event, Thread
-from typing import Callable, Iterator, Optional
+from typing import Callable, Iterator
 
 from isar.config.settings import settings
 from isar.services.utilities.mqtt_utilities import (
@@ -24,7 +24,7 @@
 from robot_interface.telemetry.mqtt_client import MqttClientInterface
 
 
-def get_next_task(task_iterator: Iterator[TASKS]) -> Optional[TASKS]:
+def get_next_task(task_iterator: Iterator[TASKS]) -> TASKS | None:
     try:
         return next(task_iterator)
     except StopIteration:
@@ -70,13 +70,13 @@ def __init__(
         self.mission_id: str = mission.id
         self.tasks = mission.tasks
 
-        self.error_message: Optional[ErrorMessage] = None
+        self.error_message: ErrorMessage | None = None
 
         Thread.__init__(self, name="Robot mission monitoring thread")
 
     def _get_task_status(self, task_id: str) -> TaskStatus:
         task_status: TaskStatus = TaskStatus.NotStarted
-        failed_task_error: Optional[ErrorMessage] = None
+        failed_task_error: ErrorMessage | None = None
         request_status_failure_counter: int = 0
 
         while (
@@ -137,7 +137,7 @@ def _get_task_status(self, task_id: str) -> TaskStatus:
 
     def _get_mission_status(self, mission_id: str) -> MissionStatus:
         mission_status: MissionStatus = MissionStatus.NotStarted
-        failed_mission_error: Optional[ErrorMessage] = None
+        failed_mission_error: ErrorMessage | None = None
         request_status_failure_counter: int = 0
 
         while (
@@ -225,7 +225,7 @@ def _get_mission_status_based_on_task_status(self) -> MissionStatus:
         else:
             return MissionStatus.Successful
 
-    def _get_and_handle_task_status(self, current_task: TASKS) -> Optional[TASKS]:
+    def _get_and_handle_task_status(self, current_task: TASKS) -> TASKS | None:
         try:
             new_task_status = self._get_task_status(current_task.id)
         except RobotTaskStatusException as e:
@@ -253,7 +253,7 @@ def _get_and_handle_task_status(self, current_task: TASKS) -> Optional[TASKS]:
             return next_task
         return current_task
 
-    def _handle_stopped_mission(self, current_task: Optional[TASKS]) -> None:
+    def _handle_stopped_mission(self, current_task: TASKS | None) -> None:
         if current_task is not None:
             current_task.status = TaskStatus.Cancelled
             publish_task_status(self.mqtt_publisher, current_task, self.mission_id)
@@ -267,7 +267,7 @@ def _handle_stopped_mission(self, current_task: Optional[TASKS]) -> None:
     def run(self) -> None:
 
         self.task_iterator: Iterator[TASKS] = iter(self.tasks)
-        current_task: Optional[TASKS] = get_next_task(self.task_iterator)
+        current_task: TASKS | None = get_next_task(self.task_iterator)
         current_task.status = TaskStatus.NotStarted  # type: ignore
         current_mission_status = MissionStatus.NotStarted