Convert format strings strings to f-strings (commaai#23241)

* Convert all text strings to f-strings

Reformats all the text from the old "%-formatted" and .format(...) format to the newer f-string format, as defined in PEP 498. This requires Python 3.6+.

Flynt 0.69 was used to reformat the strings. 120 f-strings were created in 51 files.

F-strings are in general more readable, concise and performant. See also: https://www.python.org/dev/peps/pep-0498/#rationale

* revert pyextra changes

* revert ublox.py

Co-authored-by: Willem Melching <willem.melching@gmail.com>

common/ffi_wrapper.py

@@ -28,7 +28,7 @@ def ffi_wrap(name, c_code, c_header, tmpdir="/tmp/ccache", cflags="", libraries=
     try:
       mod = __import__(cache)
     except Exception:
-      print("cache miss {0}".format(cache))
+      print(f"cache miss {cache}")
       compile_code(cache, c_code, c_header, tmpdir, cflags, libraries)
       mod = __import__(cache)
   finally:

common/file_helpers.py

@@ -35,7 +35,7 @@ def get_tmpdir_on_same_filesystem(path):
   if len(parts) > 1 and parts[1] == "scratch":
     return "/scratch/tmp"
   elif len(parts) > 2 and parts[2] == "runner":
-    return "/{}/runner/tmp".format(parts[1])
+    return f"/{parts[1]}/runner/tmp"
   return "/tmp"
 
 

common/profiler.py

@@ -42,4 +42,4 @@ def display(self):
         print("%30s: %9.2f  avg: %7.2f  percent: %3.0f   IGNORED" % (n, ms*1000.0, ms*1000.0/self.iter, ms/self.tot*100))
       else:
         print("%30s: %9.2f  avg: %7.2f  percent: %3.0f" % (n, ms*1000.0, ms*1000.0/self.iter, ms/self.tot*100))
-    print("Iter clock: %2.6f   TOTAL: %2.2f" % (self.tot/self.iter, self.tot))
+    print(f"Iter clock: {self.tot / self.iter:2.6f}   TOTAL: {self.tot:2.2f}")

common/realtime.py

@@ -79,7 +79,7 @@ def monitor_time(self) -> bool:
     remaining = self._next_frame_time - sec_since_boot()
     self._next_frame_time += self._interval
     if self._print_delay_threshold is not None and remaining < -self._print_delay_threshold:
-      print("%s lagging by %.2f ms" % (self._process_name, -remaining * 1000))
+      print(f"{self._process_name} lagging by {-remaining * 1000:.2f} ms")
       lagged = True
     self._frame += 1
     self._remaining = remaining

common/tests/test_file_helpers.py

@@ -8,7 +8,7 @@
 
 class TestFileHelpers(unittest.TestCase):
   def run_atomic_write_func(self, atomic_write_func):
-    path = "/tmp/tmp{}".format(uuid4())
+    path = f"/tmp/tmp{uuid4()}"
     with atomic_write_func(path) as f:
       f.write("test")
 

common/timeout.py

@@ -12,7 +12,7 @@ class Timeout:
   """
   def __init__(self, seconds, error_msg=None):
     if error_msg is None:
-      error_msg = 'Timed out after {} seconds'.format(seconds)
+      error_msg = f'Timed out after {seconds} seconds'
     self.seconds = seconds
     self.error_msg = error_msg
 

selfdrive/camerad/test/test_camerad.py

@@ -54,7 +54,7 @@ def test_frame_packets(self):
           self.assertTrue(abs(dfid - 1) <= SKIP_FRAME_TOLERANCE, "%s frame id diff is %d" % (camera, dfid))
 
           dts = ct - last_ts[camera]
-          self.assertTrue(abs(dts - (1000/CAMERAS[camera])) < LAG_FRAME_TOLERANCE, "%s frame t(ms) diff is %f" % (camera, dts))
+          self.assertTrue(abs(dts - (1000/CAMERAS[camera])) < LAG_FRAME_TOLERANCE, f"{camera} frame t(ms) diff is {dts:f}")
 
           last_frame_id[camera] = sm[camera].frameId
           last_ts[camera] = ct

selfdrive/car/car_helpers.py

@@ -39,7 +39,7 @@ def get_one_can(logcan):
 def load_interfaces(brand_names):
   ret = {}
   for brand_name in brand_names:
-    path = ('selfdrive.car.%s' % brand_name)
+    path = f'selfdrive.car.{brand_name}'
     CarInterface = __import__(path + '.interface', fromlist=['CarInterface']).CarInterface
 
     if os.path.exists(BASEDIR + '/' + path.replace('.', '/') + '/carstate.py'):
@@ -65,7 +65,7 @@ def _get_interface_names():
   for car_folder in [x[0] for x in os.walk(BASEDIR + '/selfdrive/car')]:
     try:
       brand_name = car_folder.split('/')[-1]
-      model_names = __import__('selfdrive.car.%s.values' % brand_name, fromlist=['CAR']).CAR
+      model_names = __import__(f'selfdrive.car.{brand_name}.values', fromlist=['CAR']).CAR
       model_names = [getattr(model_names, c) for c in model_names.__dict__.keys() if not c.startswith("__")]
       brand_names[brand_name] = model_names
     except (ImportError, IOError):

selfdrive/car/fingerprints.py

@@ -11,7 +11,7 @@ def get_attr_from_cars(attr, result=dict, combine_brands=True):
   for car_folder in [x[0] for x in os.walk(BASEDIR + '/selfdrive/car')]:
     try:
       car_name = car_folder.split('/')[-1]
-      values = __import__('selfdrive.car.%s.values' % car_name, fromlist=[attr])
+      values = __import__(f'selfdrive.car.{car_name}.values', fromlist=[attr])
       if hasattr(values, attr):
         attr_values = getattr(values, attr)
       else:

selfdrive/car/fw_versions.py

@@ -362,7 +362,7 @@ def get_fw_versions(logcan, sendcan, bus, extra=None, timeout=0.1, debug=False,
   print("Getting vin...")
   addr, vin = get_vin(logcan, sendcan, 1, retry=10, debug=args.debug)
   print(f"VIN: {vin}")
-  print("Getting VIN took %.3f s" % (time.time() - t))
+  print(f"Getting VIN took {time.time() - t:.3f} s")
   print()
 
   t = time.time()
@@ -379,4 +379,4 @@ def get_fw_versions(logcan, sendcan, bus, extra=None, timeout=0.1, debug=False,
 
   print()
   print("Possible matches:", candidates)
-  print("Getting fw took %.3f s" % (time.time() - t))
+  print(f"Getting fw took {time.time() - t:.3f} s")

selfdrive/car/honda/interface.py

@@ -301,7 +301,7 @@ def get_params(candidate, fingerprint=gen_empty_fingerprint(), car_fw=[]):  # py
       ret.lateralTuning.pid.kpV, ret.lateralTuning.pid.kiV = [[0.6], [0.18]] # TODO: can probably use some tuning
 
     else:
-      raise ValueError("unsupported car %s" % candidate)
+      raise ValueError(f"unsupported car {candidate}")
 
     # These cars use alternate user brake msg (0x1BE)
     if candidate in HONDA_BOSCH_ALT_BRAKE_SIGNAL:

selfdrive/car/tests/test_car_interfaces.py

@@ -59,7 +59,7 @@ def test_car_interfaces(self, car_name):
       car_interface.apply(CC)
 
     # Test radar interface
-    RadarInterface = importlib.import_module('selfdrive.car.%s.radar_interface' % car_params.carName).RadarInterface
+    RadarInterface = importlib.import_module(f'selfdrive.car.{car_params.carName}.radar_interface').RadarInterface
     radar_interface = RadarInterface(car_params)
     assert radar_interface
 

selfdrive/car/volkswagen/interface.py

@@ -152,7 +152,7 @@ def get_params(candidate, fingerprint=gen_empty_fingerprint(), car_fw=None):
       ret.wheelbase = 2.84
 
     else:
-      raise ValueError("unsupported car %s" % candidate)
+      raise ValueError(f"unsupported car {candidate}")
 
     ret.rotationalInertia = scale_rot_inertia(ret.mass, ret.wheelbase)
     ret.centerToFront = ret.wheelbase * 0.45

selfdrive/controls/lib/longitudinal_mpc_lib/long_mpc.py

@@ -377,8 +377,7 @@ def run(self):
     if self.solution_status != 0:
       if t > self.last_cloudlog_t + 5.0:
         self.last_cloudlog_t = t
-        cloudlog.warning("Long mpc reset, solution_status: %s" % (
-                          self.solution_status))
+        cloudlog.warning(f"Long mpc reset, solution_status: {self.solution_status}")
       self.reset()
 
 

selfdrive/controls/lib/radar_helpers.py

@@ -146,7 +146,7 @@ def get_RadarState_from_vision(self, lead_msg, v_ego):
     }
 
   def __str__(self):
-    ret = "x: %4.1f  y: %4.1f  v: %4.1f  a: %4.1f" % (self.dRel, self.yRel, self.vRel, self.aLeadK)
+    ret = f"x: {self.dRel:4.1f}  y: {self.yRel:4.1f}  v: {self.vRel:4.1f}  a: {self.aLeadK:4.1f}"
     return ret
 
   def potential_low_speed_lead(self, v_ego):

selfdrive/controls/radard.py

@@ -189,7 +189,7 @@ def radard_thread(sm=None, pm=None, can_sock=None):
 
   # import the radar from the fingerprint
   cloudlog.info("radard is importing %s", CP.carName)
-  RadarInterface = importlib.import_module('selfdrive.car.%s.radar_interface' % CP.carName).RadarInterface
+  RadarInterface = importlib.import_module(f'selfdrive.car.{CP.carName}.radar_interface').RadarInterface
 
   # *** setup messaging
   if can_sock is None:

selfdrive/controls/tests/test_alerts.py

@@ -70,7 +70,7 @@ def test_alert_text_length(self):
 
         font = fonts[alert.alert_size][i]
         w, _ = draw.textsize(txt, font)
-        msg = "type: %s msg: %s" % (alert.alert_type, txt)
+        msg = f"type: {alert.alert_type} msg: {txt}"
         self.assertLessEqual(w, max_text_width, msg=msg)
 
   def test_alert_sanity_check(self):

selfdrive/debug/can_printer.py

@@ -22,7 +22,7 @@ def can_printer(bus, max_msg, addr):
 
     if sec_since_boot() - lp > 0.1:
       dd = chr(27) + "[2J"
-      dd += "%5.2f\n" % (sec_since_boot() - start)
+      dd += f"{sec_since_boot() - start:5.2f}\n"
       for addr in sorted(msgs.keys()):
         a = msgs[addr][-1].decode('ascii', 'backslashreplace')
         x = binascii.hexlify(msgs[addr][-1]).decode('ascii')

selfdrive/debug/check_freq.py

@@ -42,6 +42,6 @@
       for name in socket_names:
         dts = np.diff(rcv_times[name])
         mean = np.mean(dts)
-        print("%s: Freq %.2f Hz, Min %.2f%%, Max %.2f%%, valid " % (name, 1.0 / mean, np.min(dts) / mean * 100, np.max(dts) / mean * 100), all(valids[name]))
+        print(f"{name}: Freq {1.0 / mean:.2f} Hz, Min {np.min(dts) / mean * 100:.2f}%, Max {np.max(dts) / mean * 100:.2f}%, valid ", all(valids[name]))
 
       prev_print = t

selfdrive/debug/cpu_usage_stat.py

@@ -110,10 +110,10 @@ def get_arg_parser():
           stat['avg'][name] = (stat['avg'][name] * (i - c) + avg * c) / (i)
           stat['cpu_samples'][name] = []
 
-        msg = 'avg: {1:.2%}, min: {2:.2%}, max: {3:.2%} {0}'.format(os.path.basename(k), stat['avg']['total'], stat['min']['total'], stat['max']['total'])
+        msg = f"avg: {stat['avg']['total']:.2%}, min: {stat['min']['total']:.2%}, max: {stat['max']['total']:.2%} {os.path.basename(k)}"
         if args.detailed_times:
           for stat_type in ['avg', 'min', 'max']:
-            msg += '\n {}: {}'.format(stat_type, [name + ':' + str(round(stat[stat_type][name]*100, 2)) for name in cpu_time_names])
+            msg += f"\n {stat_type}: {[(name + ':' + str(round(stat[stat_type][name] * 100, 2))) for name in cpu_time_names]}"
         l.append((os.path.basename(k), stat['avg']['total'], msg))
       l.sort(key=lambda x: -x[1])
       for x in l:

selfdrive/debug/dump.py

@@ -54,13 +54,13 @@
         elif args.dump_json:
           print(json.dumps(evt.to_dict()))
         elif values:
-          print("logMonotime = {}".format(evt.logMonoTime))
+          print(f"logMonotime = {evt.logMonoTime}")
           for value in values:
             if hasattr(evt, value[0]):
               item = evt
               for key in value:
                 item = getattr(item, key)
-              print("{} = {}".format(".".join(value), item))
+              print(f"{'.'.join(value)} = {item}")
           print("")
         else:
           try:

selfdrive/debug/get_fingerprint.py

@@ -27,5 +27,5 @@
 
   fingerprint = ', '.join("%d: %d" % v for v in sorted(msgs.items()))
 
-  print("number of messages {0}:".format(len(msgs)))
-  print("fingerprint {0}".format(fingerprint))
+  print(f"number of messages {len(msgs)}:")
+  print(f"fingerprint {fingerprint}")

selfdrive/debug/internal/power_monitor.py

@@ -45,7 +45,7 @@ def average(avg, sample):
       capacity_average = average(capacity_average, capacity)
       bat_temp_average = average(bat_temp_average, bat_temp)
 
-      print("%.2f volts %12.2f ma %12.2f mW %8.2f%% battery %8.1f degC" % (voltage, current, power, capacity, bat_temp))
+      print(f"{voltage:.2f} volts {current:12.2f} ma {power:12.2f} mW {capacity:8.2f}% battery {bat_temp:8.1f} degC")
       time.sleep(0.1)
   finally:
     stop_time = datetime.now()
@@ -55,8 +55,8 @@ def average(avg, sample):
     power = power_average[0]
     capacity = capacity_average[0]
     bat_temp = bat_temp_average[0]
-    print("%.2f volts %12.2f ma %12.2f mW %8.2f%% battery %8.1f degC" % (voltage, current, power, capacity, bat_temp))
-    print("  {:.2f} Seconds     {} samples".format((stop_time-start_time).total_seconds(), voltage_average[1]))
+    print(f"{voltage:.2f} volts {current:12.2f} ma {power:12.2f} mW {capacity:8.2f}% battery {bat_temp:8.1f} degC")
+    print(f"  {(stop_time - start_time).total_seconds():.2f} Seconds     {voltage_average[1]} samples")
     print("----------------------------------------------------------------")
 
     # reenable charging

selfdrive/debug/live_cpu_and_temp.py

@@ -71,7 +71,7 @@ def proc_name(proc):
       total_times = total_times_new[:]
       busy_times = busy_times_new[:]
 
-      print("CPU %.2f%% - RAM: %.2f%% - Temp %.2fC" % (100. * mean(cores), last_mem, last_temp))
+      print(f"CPU {100.0 * mean(cores):.2f}% - RAM: {last_mem:.2f}% - Temp {last_temp:.2f}C")
 
       if args.cpu and prev_proclog is not None:
         procs = {}

selfdrive/hardware/tici/power_monitor.py

@@ -40,14 +40,14 @@ def average(avg, sample):
       power_average = average(power_average, power)
       power_total_average = average(power_total_average, power_total)
 
-      print("%12.2f mW %12.2f mW %12.2f mW" % (power, power_total, power_total-power))
+      print(f"{power:12.2f} mW {power_total:12.2f} mW {power_total - power:12.2f} mW")
       time.sleep(0.25)
   finally:
     stop_time = time.monotonic()
     print("\n----------------------Average-----------------------------------")
     voltage = voltage_average[0]
     current = current_average[0]
     power = power_average[0]
-    print("%.2f volts %12.2f ma %12.2f mW %12.2f mW" % (voltage, current, power, power_total))
-    print("  {:.2f} Seconds     {} samples".format(stop_time - start_time, voltage_average[1]))
+    print(f"{voltage:.2f} volts {current:12.2f} ma {power:12.2f} mW {power_total:12.2f} mW")
+    print(f"  {stop_time - start_time:.2f} Seconds     {voltage_average[1]} samples")
     print("----------------------------------------------------------------")

selfdrive/loggerd/deleter.py

@@ -27,11 +27,11 @@ def deleter_thread(exit_event):
           continue
 
         try:
-          cloudlog.info("deleting %s" % delete_path)
+          cloudlog.info(f"deleting {delete_path}")
           shutil.rmtree(delete_path)
           break
         except OSError:
-          cloudlog.exception("issue deleting %s" % delete_path)
+          cloudlog.exception(f"issue deleting {delete_path}")
       exit_event.wait(.1)
     else:
       exit_event.wait(30)

selfdrive/loggerd/tools/mark_unuploaded.py

@@ -4,5 +4,5 @@
 from selfdrive.loggerd.uploader import UPLOAD_ATTR_NAME
 
 for fn in sys.argv[1:]:
-  print("unmarking %s" % fn)
+  print(f"unmarking {fn}")
   removexattr(fn, UPLOAD_ATTR_NAME)

selfdrive/loggerd/uploader.py

@@ -140,7 +140,7 @@ def do_upload(self, key, fn):
       cloudlog.debug("upload_url v1.4 %s %s", url, str(headers))
 
       if fake_upload:
-        cloudlog.debug("*** WARNING, THIS IS A FAKE UPLOAD TO %s ***" % url)
+        cloudlog.debug(f"*** WARNING, THIS IS A FAKE UPLOAD TO {url} ***")
 
         class FakeResponse():
           def __init__(self):

selfdrive/manager/process.py

@@ -39,7 +39,7 @@ def launcher(proc, name):
     # exec the process
     mod.main()
   except KeyboardInterrupt:
-    cloudlog.warning("child %s got SIGINT" % proc)
+    cloudlog.warning(f"child {proc} got SIGINT")
   except Exception:
     # can't install the crash handler because sys.excepthook doesn't play nice
     # with threads, so catch it here.
@@ -194,7 +194,7 @@ def start(self):
       return
 
     cwd = os.path.join(BASEDIR, self.cwd)
-    cloudlog.info("starting process %s" % self.name)
+    cloudlog.info(f"starting process {self.name}")
     self.proc = Process(name=self.name, target=nativelauncher, args=(self.cmdline, cwd))
     self.proc.start()
     self.watchdog_seen = False
@@ -214,7 +214,7 @@ def __init__(self, name, module, enabled=True, persistent=False, driverview=Fals
 
   def prepare(self):
     if self.enabled:
-      cloudlog.info("preimporting %s" % self.module)
+      cloudlog.info(f"preimporting {self.module}")
       importlib.import_module(self.module)
 
   def start(self):
@@ -225,7 +225,7 @@ def start(self):
     if self.proc is not None:
       return
 
-    cloudlog.info("starting python %s" % self.module)
+    cloudlog.info(f"starting python {self.module}")
     self.proc = Process(name=self.name, target=launcher, args=(self.module, self.name))
     self.proc.start()
     self.watchdog_seen = False
@@ -260,7 +260,7 @@ def start(self):
         # process is dead
         pass
 
-    cloudlog.info("starting daemon %s" % self.name)
+    cloudlog.info(f"starting daemon {self.name}")
     proc = subprocess.Popen(['python', '-m', self.module],  # pylint: disable=subprocess-popen-preexec-fn
                                stdin=open('/dev/null', 'r'),
                                stdout=open('/dev/null', 'w'),

selfdrive/modeld/visiontest.py

@@ -62,7 +62,7 @@ def __init__(self, input_size, model_input_size, model):
       disable_model = 0
       temporal_model = 1
     else:
-      raise ValueError("Bad model name: {}".format(model))
+      raise ValueError(f"Bad model name: {model}")
 
     prevdir = os.getcwd()
     os.chdir(_visiond_dir)  # tmp hack to find kernels

selfdrive/test/openpilotci.py

@@ -9,7 +9,7 @@
 
 def get_url(route_name, segment_num, log_type="rlog"):
   ext = "hevc" if log_type in ["fcamera", "dcamera"] else "bz2"
-  return BASE_URL + "%s/%s/%s.%s" % (route_name.replace("|", "/"), segment_num, log_type, ext)
+  return BASE_URL + f"{route_name.replace('|', '/')}/{segment_num}/{log_type}.{ext}"
 
 def upload_file(path, name):
   from azure.storage.blob import BlockBlobService  # pylint: disable=import-error

selfdrive/test/process_replay/model_replay.py

@@ -31,7 +31,7 @@
 
 
 def get_log_fn(ref_commit):
-  return "%s_%s_%s.bz2" % (TEST_ROUTE, "model_tici" if TICI else "model", ref_commit)
+  return f"{TEST_ROUTE}_{'model_tici' if TICI else 'model'}_{ref_commit}.bz2"
 
 
 def replace_calib(msg, calib):