Ruff: comprehensions conventions (#29317)

ignore c408

pyproject.toml

@@ -197,8 +197,8 @@ build-backend = "poetry.core.masonry.api"
 
 # https://beta.ruff.rs/docs/configuration/#using-pyprojecttoml
 [tool.ruff]
-select = ["E", "F", "W", "PIE"]
-ignore = ["W292", "E741", "E402"]
+select = ["E", "F", "W", "PIE", "C4"]
+ignore = ["W292", "E741", "E402", "C408"]
 line-length = 160
 target-version="py311"
 exclude = [

selfdrive/boardd/pandad.py

@@ -169,7 +169,7 @@ def main() -> NoReturn:
 
       # sort pandas to have deterministic order
       pandas.sort(key=cmp_to_key(panda_sort_cmp))
-      panda_serials = list(map(lambda p: p.get_usb_serial(), pandas))
+      panda_serials = [p.get_usb_serial() for p in pandas]
 
       # log panda fw versions
       params.put("PandaSignatures", b','.join(p.get_signature() for p in pandas))

selfdrive/car/fw_versions.py

@@ -129,7 +129,7 @@ def match_fw_to_car_exact(live_fw_versions, log=True) -> Set[str]:
       if ecu_type == Ecu.debug:
         continue
 
-      if not any([found_version in expected_versions for found_version in found_versions]):
+      if not any(found_version in expected_versions for found_version in found_versions):
         invalid.add(candidate)
         break
 
@@ -208,7 +208,7 @@ def get_brand_ecu_matches(ecu_rx_addrs):
   brand_addrs = get_brand_addrs()
   brand_matches = {brand: set() for brand, _, _ in REQUESTS}
 
-  brand_rx_offsets = set((brand, r.rx_offset) for brand, _, r in REQUESTS)
+  brand_rx_offsets = {(brand, r.rx_offset) for brand, _, r in REQUESTS}
   for addr, sub_addr, _ in ecu_rx_addrs:
     # Since we can't know what request an ecu responded to, add matches for all possible rx offsets
     for brand, rx_offset in brand_rx_offsets:

selfdrive/car/hyundai/tests/test_hyundai.py

@@ -102,16 +102,16 @@ class TestHyundaiFingerprint(unittest.TestCase):
             codes |= result
 
           if ecu[0] not in DATE_FW_ECUS or car_model in NO_DATES_PLATFORMS:
-            self.assertTrue(all({date is None for _, date in codes}))
+            self.assertTrue(all(date is None for _, date in codes))
           else:
-            self.assertTrue(all({date is not None for _, date in codes}))
+            self.assertTrue(all(date is not None for _, date in codes))
 
           if car_model == CAR.HYUNDAI_GENESIS:
             raise unittest.SkipTest("No part numbers for car model")
 
           # Hyundai places the ECU part number in their FW versions, assert all parsable
           # Some examples of valid formats: b"56310-L0010", b"56310L0010", b"56310/M6300"
-          self.assertTrue(all({b"-" in code for code, _ in codes}),
+          self.assertTrue(all(b"-" in code for code, _ in codes),
                           f"FW does not have part number: {fw}")
 
   def test_platform_codes_spot_check(self):

selfdrive/car/tesla/carstate.py

@@ -78,7 +78,7 @@ class CarState(CarStateBase):
     ret.buttonEvents = buttonEvents
 
     # Doors
-    ret.doorOpen = any([(self.can_define.dv["GTW_carState"][door].get(int(cp.vl["GTW_carState"][door]), "OPEN") == "OPEN") for door in DOORS])
+    ret.doorOpen = any((self.can_define.dv["GTW_carState"][door].get(int(cp.vl["GTW_carState"][door]), "OPEN") == "OPEN") for door in DOORS)
 
     # Blinkers
     ret.leftBlinker = (cp.vl["GTW_carState"]["BC_indicatorLStatus"] == 1)

selfdrive/car/tests/test_docs.py

@@ -85,7 +85,7 @@ class TestCarDocs(unittest.TestCase):
           raise unittest.SkipTest
 
         car_part_type = [p.type for p in car.car_parts.all_parts()]
-        car_parts = [p for p in car.car_parts.all_parts()]
+        car_parts = list(car.car_parts.all_parts())
         self.assertTrue(len(car_parts) > 0, f"Need to specify car parts: {car.name}")
         self.assertTrue(car_part_type.count(PartType.connector) == 1, f"Need to specify one harness connector: {car.name}")
         self.assertTrue(car_part_type.count(PartType.mount) == 1, f"Need to specify one mount: {car.name}")

selfdrive/locationd/test/test_laikad.py

@@ -236,8 +236,8 @@ class TestLaikad(unittest.TestCase):
           has_polys = len(vals) > 0 and max([len(v) for v in vals]) > 0
           has_fix = has_fix or out_msg.gnssMeasurements.positionECEF.valid
           if len(out_msg.gnssMeasurements.ephemerisStatuses):
-            seen_chip_eph = seen_chip_eph or any([x.source == 'gnssChip' for x in out_msg.gnssMeasurements.ephemerisStatuses])
-            seen_internet_eph = seen_internet_eph or any([x.source == 'internet' for x in out_msg.gnssMeasurements.ephemerisStatuses])
+            seen_chip_eph = seen_chip_eph or any(x.source == 'gnssChip' for x in out_msg.gnssMeasurements.ephemerisStatuses)
+            seen_internet_eph = seen_internet_eph or any(x.source == 'internet' for x in out_msg.gnssMeasurements.ephemerisStatuses)
 
         self.assertTrue(has_navs or has_polys)
         self.assertTrue(has_fix)
@@ -278,7 +278,7 @@ class TestLaikad(unittest.TestCase):
       # Verify cache is working for only nav by running a segment
       msg = verify_messages(logs, laikad, return_one_success=True)
       self.assertTrue(len(msg.gnssMeasurements.ephemerisStatuses))
-      self.assertTrue(any([x.source=='cache' for x in msg.gnssMeasurements.ephemerisStatuses]))
+      self.assertTrue(any(x.source=='cache' for x in msg.gnssMeasurements.ephemerisStatuses))
       self.assertIsNotNone(msg)
 
       #TODO test cache with only orbits

selfdrive/locationd/torqued.py

@@ -63,7 +63,7 @@ class PointBuckets:
   def __init__(self, x_bounds, min_points, min_points_total):
     self.x_bounds = x_bounds
     self.buckets = {bounds: NPQueue(maxlen=POINTS_PER_BUCKET, rowsize=3) for bounds in x_bounds}
-    self.buckets_min_points = {bounds: min_point for bounds, min_point in zip(x_bounds, min_points)}
+    self.buckets_min_points = dict(zip(x_bounds, min_points))
     self.min_points_total = min_points_total
 
   def bucket_lengths(self):
@@ -230,7 +230,7 @@ class TorqueEstimator:
       liveTorqueParameters.latAccelOffsetRaw = float(latAccelOffset)
       liveTorqueParameters.frictionCoefficientRaw = float(frictionCoeff)
 
-      if any([val is None or np.isnan(val) for val in [latAccelFactor, latAccelOffset, frictionCoeff]]):
+      if any(val is None or np.isnan(val) for val in [latAccelFactor, latAccelOffset, frictionCoeff]):
         cloudlog.exception("Live torque parameters are invalid.")
         liveTorqueParameters.liveValid = False
         self.reset()

selfdrive/test/fuzzy_generation.py

@@ -70,7 +70,7 @@ class FuzzyGenerator:
   def generate_struct(self, schema: capnp.lib.capnp._StructSchema, event: Optional[str] = None) -> st.SearchStrategy[Dict[str, Any]]:
     full_fill: List[str] = list(schema.non_union_fields)
     single_fill: List[str] = [event] if event else [self.draw(st.sampled_from(schema.union_fields))] if schema.union_fields else []
-    return st.fixed_dictionaries(dict((field, self.generate_field(schema.fields[field])) for field in full_fill + single_fill))
+    return st.fixed_dictionaries({field: self.generate_field(schema.fields[field]) for field in full_fill + single_fill})
 
   @classmethod
   def get_random_msg(cls, draw: DrawType, struct: capnp.lib.capnp._StructModule, real_floats: bool = False) -> Dict[str, Any]:

selfdrive/test/process_replay/process_replay.py

@@ -675,8 +675,8 @@ def _replay_multi_process(
       container.start(params_config, env_config, all_msgs, fingerprint, captured_output_store is not None)
       containers.append(container)
 
-    all_pubs = set([pub for container in containers for pub in container.pubs])
-    all_subs = set([sub for container in containers for sub in container.subs])
+    all_pubs = {pub for container in containers for pub in container.pubs}
+    all_subs = {sub for container in containers for sub in container.subs}
     lr_pubs = all_pubs - all_subs
     pubs_to_containers = {pub: [container for container in containers if pub in container.pubs] for pub in all_pubs}
 

selfdrive/test/test_onroad.py

@@ -258,7 +258,7 @@ class TestOnroad(unittest.TestCase):
         cpu_ok = False
 
     # Ensure there's no missing procs
-    all_procs = set([p.name for p in self.service_msgs['managerState'][0].managerState.processes if p.shouldBeRunning])
+    all_procs = {p.name for p in self.service_msgs['managerState'][0].managerState.processes if p.shouldBeRunning}
     for p in all_procs:
       with self.subTest(proc=p):
         assert any(p in pp for pp in PROCS.keys()), f"Expected CPU usage missing for {p}"

selfdrive/test/test_valgrind_replay.py

@@ -73,7 +73,7 @@ class TestValgrind(unittest.TestCase):
       self.leak = False
 
   def replay_process(self, config, logreader):
-    pub_sockets = [s for s in config.pub_sub.keys()]  # We dump data from logs here
+    pub_sockets = list(config.pub_sub.keys())  # We dump data from logs here
     sub_sockets = [s for _, sub in config.pub_sub.items() for s in sub]  # We get responses here
     pm = messaging.PubMaster(pub_sockets)
     sm = messaging.SubMaster(sub_sockets)

selfdrive/ui/tests/test_translations.py

@@ -88,7 +88,7 @@ class TestTranslations(unittest.TestCase):
                 continue
 
               self.assertNotIn(None, numerusform, "Ensure all plural translation forms are completed.")
-              self.assertTrue(all([re.search("%[0-9]+", t) is None for t in numerusform]),
+              self.assertTrue(all(re.search("%[0-9]+", t) is None for t in numerusform),
                               "Plural translations must use %n, not %1, %2, etc.: {}".format(numerusform))
 
   def test_no_locations(self):

system/loggerd/tests/test_loggerd.py

@@ -130,7 +130,7 @@ class TestLoggerd(unittest.TestCase):
 
     # check params
     logged_params = {entry.key: entry.value for entry in initData.params.entries}
-    expected_params = set(k for k, _, __ in fake_params) | {'LaikadEphemerisV3'}
+    expected_params = {k for k, _, __ in fake_params} | {'LaikadEphemerisV3'}
     assert set(logged_params.keys()) == expected_params, set(logged_params.keys()) ^ expected_params
     assert logged_params['LaikadEphemerisV3'] == b'', f"DONT_LOG param value was logged: {repr(logged_params['LaikadEphemerisV3'])}"
     for param_key, initData_key, v in fake_params:

system/loggerd/uploader.py

@@ -46,7 +46,7 @@ class FakeResponse:
 UploadResponse = Union[requests.Response, FakeResponse]
 
 def get_directory_sort(d: str) -> List[str]:
-  return list(map(lambda s: s.rjust(10, '0'), d.rsplit('--', 1)))
+  return [s.rjust(10, '0') for s in d.rsplit('--', 1)]
 
 def listdir_by_creation(d: str) -> List[str]:
   try:

system/sensord/rawgps/compare.py

@@ -19,8 +19,8 @@ if __name__ == "__main__":
     recv_time = report.milliseconds / 1000
 
     car = []
-    print("qcom has ", list(sorted([x.svId for x in report.sv])))
-    print("ublox has", list(sorted([x.svId for x in meas if x.gnssId == (6 if GLONASS else 0)])))
+    print("qcom has ", sorted([x.svId for x in report.sv]))
+    print("ublox has", sorted([x.svId for x in meas if x.gnssId == (6 if GLONASS else 0)]))
     for i in report.sv:
       # match to ublox
       tm = None