type hints

sunnypilot/mapd/mapd_installer.py

@@ -7,12 +7,14 @@ import traceback
 import requests
 from pathlib import Path
 from urllib.request import urlopen
-import openpilot.system.sentry as sentry
+
 from cereal import messaging
 from openpilot.common.params import Params
+from openpilot.sunnypilot.mapd.mapd_manager import MAPD_PATH, MAPD_BIN_DIR
+from openpilot.system.hardware.hw import Paths
 from openpilot.system.ui.spinner import Spinner
-from openpilot.sunnypilot.mapd.mapd_manager import COMMON_DIR, MAPD_PATH, MAPD_BIN_DIR
 from openpilot.system.version import is_prebuilt
+import openpilot.system.sentry as sentry
 
 VERSION = 'v1.9.0'
 URL = f"https://github.com/pfeiferj/openpilot-mapd/releases/download/{VERSION}/mapd"
@@ -22,28 +24,28 @@ class MapdInstallManager:
   def __init__(self, spinner_ref: Spinner):
     self._spinner = spinner_ref
 
-  def download(self):
+  def download(self) -> None:
     self.ensure_directories_exist()
     self._download_file()
     self.update_installed_version(VERSION)
 
-  def check_and_download(self):
+  def check_and_download(self) -> None:
     if self.download_needed():
       self.download()
 
   @staticmethod
-  def download_needed():
+  def download_needed() -> bool:
     return not os.path.exists(MAPD_PATH) or MapdInstallManager.get_installed_version() != VERSION
 
   @staticmethod
-  def ensure_directories_exist():
-    if not os.path.exists(COMMON_DIR):
-      os.makedirs(COMMON_DIR)
+  def ensure_directories_exist() -> None:
+    if not os.path.exists(Paths.mapd_root()):
+      os.makedirs(Paths.mapd_root())
     if not os.path.exists(MAPD_BIN_DIR):
       os.makedirs(MAPD_BIN_DIR)
 
   @staticmethod
-  def _safe_write_and_set_executable(file_path, content):
+  def _safe_write_and_set_executable(file_path: Path, content: bytes) -> None:
     with open(file_path, 'wb') as output:
       output.write(content)
       output.flush()
@@ -51,7 +53,7 @@ class MapdInstallManager:
     current_permissions = stat.S_IMODE(os.lstat(file_path).st_mode)
     os.chmod(file_path, current_permissions | stat.S_IEXEC)
 
-  def _download_file(self, num_retries=5):
+  def _download_file(self, num_retries=5) -> None:
     temp_file = Path(MAPD_PATH + ".tmp")
     download_timeout = 60
     for cnt in range(num_retries):
@@ -75,14 +77,14 @@ class MapdInstallManager:
     logging.error("Failed to download file after all retries")
 
   @staticmethod
-  def update_installed_version(version):
+  def update_installed_version(version: str) -> None:
     Params().put("MapdVersion", version)
 
   @staticmethod
-  def get_installed_version():
+  def get_installed_version() -> str | None:
     return Params().get("MapdVersion", encoding="utf-8")
 
-  def wait_for_internet_connection(self, return_on_failure=False):
+  def wait_for_internet_connection(self, return_on_failure: bool = False) -> bool | None:
     max_retries = 10
     for retries in range(max_retries + 1):
       self._spinner.update(f"Waiting for internet connection... [{retries}/{max_retries}]")
@@ -95,7 +97,7 @@ class MapdInstallManager:
         if return_on_failure and retries == max_retries:
           return False
 
-  def non_prebuilt_install(self):
+  def non_prebuilt_install(self) -> bool | None:
     sm = messaging.SubMaster(['deviceState'])
     metered = sm['deviceState'].networkMetered