fix linting

sunnypilot/mapd/live_map_data/base_map_data.py

@@ -23,7 +23,7 @@ class BaseMapData(ABC):
     pass
 
   @abstractmethod
-  def get_next_speed_limit_and_distance(self) -> (float, float):
+  def get_next_speed_limit_and_distance(self) -> tuple[float, float]:
     pass
 
   @abstractmethod
@@ -33,7 +33,7 @@ class BaseMapData(ABC):
   def _is_gps_data_valid(self) -> bool:
     all_sock_alive = self._sub_master.all_alive(service_list=[self._gps_sock])
     all_sock_valid = self._sub_master.all_valid(service_list=[self._gps_sock])
-    return all_sock_alive and all_sock_valid
+    return bool(all_sock_alive and all_sock_valid)
 
   def get_current_location(self) -> Coordinate | None:
     self._gps_sock = "liveLocationKalman"
@@ -47,7 +47,7 @@ class BaseMapData(ABC):
 
     kalman_bearing_deg = math.degrees(_last_gps.calibratedOrientationNED.value[2])
     kalman_speed = _last_gps.velocityCalibrated.value[2]
-    kalman_latitude = _last_gps.positionGeodetic.value[0];
+    kalman_latitude = _last_gps.positionGeodetic.value[0]
     kalman_longitude = _last_gps.positionGeodetic.value[1]
 
     result = Coordinate(kalman_latitude, kalman_longitude)
@@ -97,7 +97,9 @@ class BaseMapData(ABC):
     )
 
     self._pub_master.send('liveMapDataSP', live_map_data_sp)
-    get_debug(f"SRC: [{self.__class__.__name__}] | SLC: [{speed_limit}] | NSL: [{next_speed_limit}] | NSLD: [{next_speed_limit_distance}] | CRN: [{current_road_name}] | GPS: [{self._last_gps}] Annotations: [{', '.join(f'{key}: {value}' for key, value in self._last_gps.annotations.items()) if self._last_gps else []}]")
+    get_debug(f"SRC: [{self.__class__.__name__}] | SLC: [{speed_limit}] | NSL: [{next_speed_limit}] | " +
+              f"NSLD: [{next_speed_limit_distance}] | CRN: [{current_road_name}] | GPS: [{self._last_gps}] " +
+              f"Annotations: [{', '.join(f'{key}: {value}' for key, value in self._last_gps.annotations.items()) if self._last_gps else []}]")
 
   def tick(self):
     self._sub_master.update()

sunnypilot/mapd/live_map_data/osm_map_data.py

@@ -35,13 +35,13 @@ class OsmMapData(BaseMapData):
     current_road_name = self.mem_params.get("RoadName", encoding='utf8')
     return current_road_name if current_road_name else ""
 
-  def get_next_speed_limit_and_distance(self):
+  def get_next_speed_limit_and_distance(self) -> tuple[float, float]:
     next_speed_limit_section_str = self.mem_params.get("NextMapSpeedLimit", encoding='utf8')
     next_speed_limit_section = json.loads(next_speed_limit_section_str) if next_speed_limit_section_str else {}
     next_speed_limit = next_speed_limit_section.get('speedlimit', 0.0)
     next_speed_limit_latitude = next_speed_limit_section.get('latitude')
     next_speed_limit_longitude = next_speed_limit_section.get('longitude')
-    next_speed_limit_distance = 0
+    next_speed_limit_distance = 0.0
 
     if next_speed_limit_latitude and next_speed_limit_longitude:
       next_speed_limit_coordinates = Coordinate(next_speed_limit_latitude, next_speed_limit_longitude)

sunnypilot/mapd/mapd_installer.py

@@ -9,8 +9,8 @@ from pathlib import Path
 from urllib.request import urlopen
 import openpilot.system.sentry as sentry
 from cereal import messaging
-from common.spinner import Spinner
-from common.params import Params
+from openpilot.common.spinner import Spinner
+from openpilot.common.params import Params
 from openpilot.system.mapd_manager import COMMON_DIR, MAPD_PATH, MAPD_BIN_DIR
 from openpilot.system.version import is_prebuilt
 
@@ -100,7 +100,7 @@ class MapdInstallManager:
     metered = sm['deviceState'].networkMetered
 
     if metered:
-      self._spinner.update(f"Can't proceed with mapd install since network is metered!")
+      self._spinner.update("Can't proceed with mapd install since network is metered!")
       time.sleep(5)
       return False
 
@@ -119,8 +119,8 @@ class MapdInstallManager:
 
     except Exception:
       for i in range(6):
-        self._spinner.update(f"Failed to download OSM maps won't work until properly downloaded!"
-                             f"Try again manually rebooting. "
+        self._spinner.update("Failed to download OSM maps won't work until properly downloaded!" +
+                             "Try again manually rebooting. " +
                              f"Boot will continue in {5 - i}s...")
         time.sleep(1)
 

sunnypilot/mapd/mapd_manager.py

@@ -80,7 +80,7 @@ def _request_refresh_osm_bounds_data(self):
   mem_params.put("OSMDownloadBounds", json.dumps(current_bounding_box))
 
 
-def request_refresh_osm_location_data(nations: [str], states: [str] = None):
+def request_refresh_osm_location_data(nations: list[str], states: list[str] = None) -> None:
   params.put("OsmDownloadedDate", str(time.time()))
   params.put_bool("OsmDbUpdatesCheck", False)
 
@@ -93,7 +93,7 @@ def request_refresh_osm_location_data(nations: [str], states: [str] = None):
   mem_params.put("OSMDownloadLocations", osm_download_locations)
 
 
-def filter_nations_and_states(nations: [str], states: [str] = None):
+def filter_nations_and_states(nations: list[str], states: list[str] = None):
   """Filters and prepares nation and state data for OSM map download.
 
   If the nation is 'US' and a specific state is provided, the nation 'US' is removed from the list.

sunnypilot/selfdrive/controls/lib/speed_limit_controller/speed_limit_controller.py

@@ -6,7 +6,7 @@ from openpilot.common.params import Params
 from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit_controller import LIMIT_PERC_OFFSET_BP, LIMIT_PERC_OFFSET_V, \
   PARAMS_UPDATE_PERIOD, TEMP_INACTIVE_GUARD_PERIOD, LIMIT_SPEED_OFFSET_TH, EventName, SpeedLimitControlState
 from openpilot.selfdrive.controls.lib.drive_helpers import CONTROL_N
-from openpilot.selfdrive.selfdrived.events import Events, ET
+from openpilot.selfdrive.selfdrived.events import ET
 from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit_controller.common import Source, Policy, Engage, OffsetType
 from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit_controller.helpers import description_for_state, debug
 from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit_controller.speed_limit_resolver import SpeedLimitResolver
@@ -150,7 +150,8 @@ class SpeedLimitController:
       self._is_metric = self._params.get_bool("IsMetric")
       self._ms_to_local = CV.MS_TO_KPH if self._is_metric else CV.MS_TO_MPH
       self._resolver.change_policy(self._policy)
-      self._engage_type = Engage.auto if self._pcm_cruise_op_long else np.clip(int(self._params.get("SpeedLimitEngageType", encoding='utf8')), Engage.auto, Engage.user_confirm)
+      self._engage_type = Engage.auto if self._pcm_cruise_op_long else \
+                          np.clip(int(self._params.get("SpeedLimitEngageType", encoding='utf8')), Engage.auto, Engage.user_confirm)
       debug(f'Updated Speed limit params. enabled: {self._is_enabled}, with offset: {self._offset_type}')
       self._last_params_update = self._current_time
 
@@ -176,8 +177,9 @@ class SpeedLimitController:
 
     self._v_cruise_rounded = int(round(self._v_cruise_setpoint * self._ms_to_local))
     self._v_cruise_prev_rounded = int(round(self._v_cruise_setpoint_prev * self._ms_to_local))
-    self._speed_limit_offsetted_rounded = int(0) if self._speed_limit == 0 else int(round((self._speed_limit + self.speed_limit_offset) * self._ms_to_local))
-    self._speed_limit_warning_offsetted_rounded = int(0) if self._speed_limit == 0 else int(round((self._speed_limit + self.speed_limit_warning_offset) * self._ms_to_local))
+    self._speed_limit_offsetted_rounded = 0 if self._speed_limit == 0 else int(round((self._speed_limit + self.speed_limit_offset) * self._ms_to_local))
+    self._speed_limit_warning_offsetted_rounded = 0 if self._speed_limit == 0 else \
+                                                  int(round((self._speed_limit + self.speed_limit_warning_offset) * self._ms_to_local))
 
   def transition_state_from_inactive(self):
     """ Make state transition from inactive state """
@@ -294,7 +296,7 @@ class SpeedLimitController:
         elif self._speed_limit_changed != 0:
           events.add(EventName.speedLimitValueChange)
 
-  def update(self, enabled, v_ego, a_ego, sm, v_cruise_setpoint, events=Events()):
+  def update(self, enabled, v_ego, a_ego, sm, v_cruise_setpoint, events):
     _car_state = sm['carState']
     self._op_enabled = enabled and sm['controlsState'].enabled and _car_state.cruiseState.enabled and \
                        not (_car_state.brakePressed and (not self._brake_pressed_prev or not _car_state.standstill)) and \

sunnypilot/selfdrive/controls/lib/speed_limit_controller/speed_limit_resolver.py

@@ -120,13 +120,13 @@ class SpeedLimitResolver:
       # They are ordered in the order of preference, so we pick the first that's non zero
       for source in sources_for_policy:
         if self._limit_solutions[source] > 0.:
-          return source
+          return Source(source)
 
     limits = np.array([self._limit_solutions[source] for source in sources_for_policy], dtype=float)
     sources = np.array([source.value for source in sources_for_policy], dtype=int)
 
     if len(limits) > 0:
       min_idx = np.argmin(limits)
-      return sources[min_idx]
+      return Source(sources[min_idx])
 
     return None

sunnypilot/selfdrive/controls/lib/speed_limit_controller/tests/test_speed_limit_resolver.py

@@ -2,36 +2,35 @@ import random
 import time
 
 import pytest
-from unittest.mock import MagicMock
+from pytest_mock import MockerFixture
 
-from openpilot.selfdrive.controls.lib.drive_helpers import LIMIT_MAX_MAP_DATA_AGE
+from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit_controller import LIMIT_MAX_MAP_DATA_AGE
 
 # from selfdrive.controls.lib.speed_limit_controller_tbd import SpeedLimitResolver as OriginalSpeedLimitResolver
-from selfdrive.controls.lib.sunnypilot.speed_limit_resolver import SpeedLimitResolver as RefactoredSpeedLimitResolver
-from openpilot.selfdrive.controls.lib.sunnypilot.common import Source
-from openpilot.selfdrive.controls.lib.sunnypilot.common import Policy
+from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit_controller.speed_limit_resolver import SpeedLimitResolver as RefactoredSpeedLimitResolver
+from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit_controller.common import Source, Policy
 
 
-def create_mock(properties):
-  mock = MagicMock()
-  for property, value in properties.items():
-    setattr(mock, property, value)
+def create_mock(properties, mocker: MockerFixture):
+  mock = mocker.MagicMock()
+  for _property, value in properties.items():
+    setattr(mock, _property, value)
   return mock
 
 
-def setup_sm_mock():
+def setup_sm_mock(mocker: MockerFixture):
   cruise_speed_limit = random.uniform(0, 120)
   nav_instruction_limit = random.uniform(0, 120)
   live_map_data_limit = random.uniform(0, 120)
 
-  cruise_state = create_mock({'speedLimit': cruise_speed_limit})
+  cruise_state = create_mock({'speedLimit': cruise_speed_limit}, mocker)
   car_state = create_mock({
     'cruiseState': cruise_state,
     'gasPressed': False,
     'brakePressed': False,
     'standstill': False,
-  })
-  nav_instruction = create_mock({'speedLimit': nav_instruction_limit})
+  }, mocker)
+  nav_instruction = create_mock({'speedLimit': nav_instruction_limit}, mocker)
   live_map_data = create_mock({
     'speedLimit': live_map_data_limit,
     'speedLimitValid': True,
@@ -39,8 +38,8 @@ def setup_sm_mock():
     'speedLimitAheadValid': 0.,
     'speedLimitAheadDistance': 0.,
     'lastGpsTimestamp': time.time() * 1e3,
-  })
-  sm_mock = MagicMock()
+  }, mocker)
+  sm_mock = mocker.MagicMock()
   sm_mock.__getitem__.side_effect = lambda key: {
     'carState': car_state,
     'navInstruction': nav_instruction,
@@ -74,9 +73,9 @@ class TestSpeedLimitResolverValidation:
         assert resolver._distance_solutions[source] == 0.
 
   @parametrized_policies
-  def test_resolver(self, resolver_class, policy, sm_key, function_key):
+  def test_resolver(self, resolver_class, policy, sm_key, function_key, mocker: MockerFixture):
     resolver = resolver_class(policy)
-    sm_mock = setup_sm_mock()
+    sm_mock = setup_sm_mock(mocker)
     source_speed_limit = sm_mock[sm_key].cruiseState.speedLimit if sm_key == 'carState' else sm_mock[sm_key].speedLimit
 
     # Assert the resolver
@@ -84,9 +83,9 @@ class TestSpeedLimitResolverValidation:
     assert speed_limit == source_speed_limit
     assert source == Source[function_key]
 
-  def test_resolver_combined(self, resolver_class):
+  def test_resolver_combined(self, resolver_class, mocker: MockerFixture):
     resolver = resolver_class(Policy.combined)
-    sm_mock = setup_sm_mock()
+    sm_mock = setup_sm_mock(mocker)
     socket_to_source = {'carState': Source.car_state, 'liveMapDataSP': Source.map_data, 'navInstruction': Source.nav}
     minimum_key, minimum_speed_limit = min(
       ((key, sm_mock[key].cruiseState.speedLimit) if key == 'carState' else (key, sm_mock[key].speedLimit) for key in
@@ -98,9 +97,9 @@ class TestSpeedLimitResolverValidation:
     assert source == socket_to_source[minimum_key]
 
   @parametrized_policies
-  def test_parser(self, resolver_class, policy, sm_key, function_key):
+  def test_parser(self, resolver_class, policy, sm_key, function_key, mocker: MockerFixture):
     resolver = resolver_class(policy)
-    sm_mock = setup_sm_mock()
+    sm_mock = setup_sm_mock(mocker)
     source_speed_limit = sm_mock[sm_key].cruiseState.speedLimit if sm_key == 'carState' else sm_mock[sm_key].speedLimit
 
     # Assert the parsing
@@ -109,11 +108,11 @@ class TestSpeedLimitResolverValidation:
     assert resolver._distance_solutions[Source[function_key]] == 0.
 
   @pytest.mark.parametrize("policy", list(Policy), ids=lambda policy: policy.name)
-  def test_resolve_interaction_in_update(self, resolver_class, policy):
+  def test_resolve_interaction_in_update(self, resolver_class, policy, mocker: MockerFixture):
     v_ego = 50
     resolver = resolver_class(policy)
 
-    sm_mock = setup_sm_mock()
+    sm_mock = setup_sm_mock(mocker)
     _speed_limit, _distance, _source = resolver.resolve(v_ego, 0, sm_mock)
 
     # After resolution
@@ -122,9 +121,9 @@ class TestSpeedLimitResolverValidation:
     assert _source is not None
 
   @pytest.mark.parametrize("policy", list(Policy), ids=lambda policy: policy.name)
-  def test_old_map_data_ignored(self, resolver_class, policy):
+  def test_old_map_data_ignored(self, resolver_class, policy, mocker: MockerFixture):
     resolver = resolver_class(policy)
-    sm_mock = MagicMock()
+    sm_mock = mocker.MagicMock()
     sm_mock['liveMapDataSP'].lastGpsTimestamp = (time.time() - 2 * LIMIT_MAX_MAP_DATA_AGE) * 1e3
     resolver._sm = sm_mock
     resolver._get_from_map_data()

system/hardware/hw.py

@@ -77,3 +77,11 @@ class Paths:
       return str(Path(Paths.comma_home()) / "community" / "crashes")
     else:
       return "/data/community/crashes"
+
+
+  @staticmethod
+  def mapd_root() -> str:
+    if PC:
+      return str(Path(Paths.comma_home()) / "media" / "0" / "osm")
+    else:
+      return "/data/media/0/osm"