Merge branch 'johan/f-strings' into python

.pylintrc

@@ -1,6 +1,5 @@
 [MESSAGES CONTROL]
 disable=
-    consider-using-f-string, # We should, in a separate change
     consider-using-max-builtin,
     disallowed-name,
     fixme,

devbin/benchmark_ipcmap.py

@@ -47,7 +47,7 @@ def get_timings(file, pid):
     """
     t0 = time.time()
     files = None
-    with open(file, "r", encoding="utf-8") as lsof_output:
+    with open(file, encoding="utf-8") as lsof_output:
         files = px_file.lsof_to_files(lsof_output.read())
     t1 = time.time()
     dt_load = t1 - t0
@@ -65,16 +65,16 @@ def print_statistics(name, values):
     highest = max(values)
     middle = (lowest + highest) / 2
     radius = (highest - lowest) / 2
-    print("{} is {:.2f}s±{:.2f}s".format(name, middle, radius))
+    print(f"{name} is {middle:.2f}s±{radius:.2f}s")
 
 
 def main(lsof_file):
     print("Finding most popular PID...")
     files = None
-    with open(lsof_file, "r", encoding="utf-8") as lsof_output:
+    with open(lsof_file, encoding="utf-8") as lsof_output:
         files = px_file.lsof_to_files(lsof_output.read())
     pid = get_most_common_pid(files)
-    print("Most popular PID: {}".format(pid))
+    print(f"Most popular PID: {pid}")
 
     end = time.time() + DURATION_S
     lap_number = 0
@@ -83,7 +83,7 @@ def main(lsof_file):
     total_times = []
     while time.time() < end:
         lap_number += 1
-        print("Lap {}, {:.0f}s left...".format(lap_number, end - time.time()))
+        print(f"Lap {lap_number}, {end - time.time():.0f}s left...")
         load_time, mapping_time = get_timings(lsof_file, pid)
         load_times.append(load_time)
         mapping_times.append(mapping_time)

devbin/benchmark_proc_get_all.py

@@ -26,7 +26,7 @@ def main():
     t1 = time.time()
     dt_seconds = t1 - t0
 
-    print("Getting all processes takes {:.0f}ms".format(1000 * dt_seconds / LAPS))
+    print(f"Getting all processes takes {1000 * dt_seconds / LAPS:.0f}ms")
 
 
 if __name__ == "__main__":

px/px.py

@@ -229,7 +229,7 @@ def _main(argv: List[str]) -> None:
         processes = px_process.get_all()
         process = px_processinfo.find_process_by_pid(pid, processes)
         if not process:
-            sys.exit("No such PID: {}\n".format(pid))
+            sys.exit(f"No such PID: {pid}")
 
         px_pager.page_process_info(process, processes)
         return

px/px_commandline.py

@@ -312,9 +312,7 @@ def get_java_command(commandline: str) -> str:
                 return java
             return prettify_fully_qualified_java_class(component)
 
-        raise ValueError(
-            "Unhandled state <{}> at <{}> for: {}".format(state, component, array)
-        )
+        raise ValueError(f"Unhandled state <{state}> at <{component}> for: {array}")
 
     # We got to the end without being able to come up with a better name, give up
     return java

px/px_file.py

@@ -155,9 +155,7 @@ def build(self) -> PxFile:
         return pxFile
 
     def __repr__(self):
-        return "PxFileBuilder(pid={}, name={}, type={})".format(
-            self.pid, self.name, self.type
-        )
+        return f"PxFileBuilder(pid={self.pid}, name={self.name}, type={self.type})"
 
 
 def resolve_endpoint(endpoint: str) -> str:
@@ -265,9 +263,7 @@ def lsof_to_files(lsof: str) -> List[PxFile]:
             file_builder.inode = value
 
         else:
-            raise Exception(
-                "Unhandled type <{}> for shard <{}>".format(infotype, shard)
-            )
+            raise Exception(f"Unhandled type <{infotype}> for shard <{shard}>")
 
     if file_builder:
         # Don't forget the last file

px/px_install.py

@@ -13,10 +13,10 @@ def install(src, dest):
     try:
         _install(src, dest)
     except Exception as e:  # pylint: disable=broad-except
-        sys.stderr.write("Installing {} failed, please retry with sudo\n".format(dest))
-        sys.stderr.write("Error was: {}\n".format(str(e)))
+        sys.stderr.write(f"Installing {dest} failed, please retry with sudo\n")
+        sys.stderr.write(f"Error was: {str(e)}\n")
         sys.exit(1)
-    print("Created: {}".format(dest))
+    print(f"Created: {dest}")
 
 
 def _install(src, dest):
@@ -30,22 +30,22 @@ def _install(src, dest):
         return
 
     if not os.path.isfile(src):
-        raise IOError("Source is not a file: %s" % (src,))
+        raise OSError(f"Source is not a file: {src}")
 
     parent = os.path.dirname(dest)
     if not os.path.isdir(parent):
-        raise IOError("Destination parent is not a directory: %s" % (parent,))
+        raise OSError(f"Destination parent is not a directory: {parent}")
 
     if os.path.isdir(dest):
-        raise IOError("Destination is a directory, won't replace that: %s" % (dest,))
+        raise OSError(f"Destination is a directory, won't replace that: {dest}")
 
     # Make sure nothing's in the way
     try:
         os.remove(dest)
     except OSError:
         pass
     if os.path.exists(dest):
-        raise IOError("Can't remove existing entry: %s" % (dest,))
+        raise OSError(f"Can't remove existing entry: {dest}")
 
     shutil.copyfile(src, dest)
     os.chmod(dest, 0o755)

px/px_ioload.py

@@ -56,7 +56,7 @@ def __init__(self, name: str, bytecount: int) -> None:
         self.bytecount = bytecount
 
     def __repr__(self):
-        return 'Sample[name="{}", count={}]'.format(self.name, self.bytecount)
+        return f'Sample[name="{self.name}", count={self.bytecount}]'
 
     def __eq__(self, o):
         return self.bytecount == o.bytecount and self.name == o.name
@@ -162,9 +162,7 @@ def __init__(self, throughput: float, high_watermark: float) -> None:
 
         if throughput > high_watermark:
             raise ValueError(
-                "High watermark {} lower than throughput {}".format(
-                    high_watermark, throughput
-                )
+                f"High watermark {high_watermark} lower than throughput {throughput}"
             )
 
         self.throughput = throughput
@@ -343,6 +341,8 @@ def get_load_string(self) -> str:
         current_throughput, max_throughput = px_units.bytes_to_strings(
             math.trunc(bottleneck[1]), math.trunc(bottleneck[2])
         )
+
+        # pylint: disable=consider-using-f-string
         return "[{} / {}] {}".format(
             px_terminal.bold(current_throughput + "/s"),
             max_throughput + "/s",

px/px_ipc_map.py

@@ -119,11 +119,7 @@ def _create_fds(self, is_root: bool) -> Dict[int, str]:
                     name = file.device
                 if name and name.startswith("->"):
                     name = name[2:]
-                fds[file.fd] = "[{}] <{}> ({})".format(
-                    file.type,
-                    excuse,
-                    name,
-                )
+                fds[file.fd] = f"[{file.type}] <{excuse}> ({name})"
 
         # Traverse network connections and update FDs as required
         for network_connection in self.network_connections:
@@ -148,7 +144,7 @@ def _create_fds(self, is_root: bool) -> Dict[int, str]:
                 name = link.name
                 if name and name.startswith("->"):
                     name = name[2:]
-                fds[link.fd] = "[{}] -> {} ({})".format(link.type, str(target), name)
+                fds[link.fd] = f"[{link.type}] -> {target} ({name})"
 
         return fds
 

px/px_load.py

@@ -14,7 +14,7 @@
 
 physical, logical = px_cpuinfo.get_core_count()
 physical_string = px_terminal.bold(str(physical) + " cores")
-cores_string = "[{} | {} virtual]".format(physical_string, logical)
+cores_string = f"[{physical_string} | {logical} virtual]"
 
 
 def average_to_level(average, peak):
@@ -100,7 +100,7 @@ def get_load_string(load_values: Tuple[float, float, float] = None) -> str:
 
     avg0to1, avg1to5, avg5to15 = load_values
 
-    load_string = "{:.1f}".format(avg0to1)
+    load_string = f"{avg0to1:.1f}"
     if avg0to1 <= physical:
         load_string = px_terminal.green(load_string)
     elif avg0to1 <= logical:
@@ -114,4 +114,4 @@ def get_load_string(load_values: Tuple[float, float, float] = None) -> str:
     # Increase intensity for more recent times
     graph = px_terminal.faint(graph[0:3]) + graph[3:6] + px_terminal.bold(graph[6:])
 
-    return "{}  {}  [15m history: {}]".format(load_string, cores_string, graph)
+    return f"{load_string}  {cores_string}  [15m history: {graph}]"

px/px_process.py

@@ -15,7 +15,6 @@
 
 from typing import Dict
 from typing import MutableSet
-from typing import Text
 from typing import Optional
 from typing import List
 from typing import Iterable
@@ -42,11 +41,11 @@
 TIMEZONE = dateutil.tz.tzlocal()
 
 
-uid_to_username_cache: Dict[int, Text] = {}
-get_command_cache: Dict[Text, Text] = {}
+uid_to_username_cache: Dict[int, str] = {}
+get_command_cache: Dict[str, str] = {}
 
 
-def _parse_time(time_s: Text) -> datetime.datetime:
+def _parse_time(time_s: str) -> datetime.datetime:
     """
     Parse a local date from ps into a datetime.datetime object.
 
@@ -85,11 +84,11 @@ class PxProcess:
     # pylint: disable=attribute-defined-outside-init
     def __init__(
         self,
-        cmdline: Text,
+        cmdline: str,
         pid: int,
         rss_kb: int,
-        start_time_string: Text,
-        username: Text,
+        start_time_string: str,
+        username: str,
         now: datetime.datetime,
         ppid: Optional[int],
         memory_percent: Optional[float] = None,
@@ -137,12 +136,12 @@ def __init__(
         self.memory_percent = memory_percent
         self.memory_percent_s: str = "--"
         if memory_percent is not None:
-            self.memory_percent_s = "{:.0f}%".format(memory_percent)
+            self.memory_percent_s = f"{memory_percent:.0f}%"
 
         self.cpu_percent = cpu_percent
         self.cpu_percent_s: str = "--"
         if cpu_percent is not None:
-            self.cpu_percent_s = "{:.0f}%".format(cpu_percent)
+            self.cpu_percent_s = f"{cpu_percent:.0f}%"
 
         # Setting the CPU time like this implicitly recomputes the score
         self.set_cpu_time_seconds(cpu_time)
@@ -184,7 +183,7 @@ def _recompute_score(self):
         )
 
     def set_cpu_time_seconds(self, seconds: Optional[float]) -> None:
-        self.cpu_time_s: Text = "--"
+        self.cpu_time_s: str = "--"
         self.cpu_time_seconds = None
         if seconds is not None:
             self.cpu_time_s = seconds_to_str(seconds)
@@ -257,17 +256,18 @@ def is_alive(self):
 
 class PxProcessBuilder:
     def __init__(self):
-        self.cmdline: Optional[Text] = None
+        self.cmdline: Optional[str] = None
         self.pid: Optional[int] = None
         self.ppid: Optional[int] = None
         self.rss_kb: Optional[int] = None
-        self.start_time_string: Optional[Text] = None
-        self.username: Optional[Text] = None
+        self.start_time_string: Optional[str] = None
+        self.username: Optional[str] = None
         self.cpu_percent: Optional[float] = None
         self.cpu_time: Optional[float] = None
         self.memory_percent: Optional[float] = None
 
     def __repr__(self):
+        # pylint: disable=consider-using-f-string
         return (
             "start_time_string=%r pid=%r ppid=%r user=%r cpu%%=%r cputime=%r mem%%=%r cmd=<%r>"
             % (
@@ -302,7 +302,7 @@ def build(self, now: datetime.datetime) -> PxProcess:
         )
 
 
-def parse_time(timestring: Text) -> float:
+def parse_time(timestring: str) -> float:
     """Convert a CPU time string returned by ps to a number of seconds"""
 
     match = CPUTIME_OSX.match(timestring)
@@ -329,7 +329,7 @@ def parse_time(timestring: Text) -> float:
     raise ValueError("Unparsable timestamp: <" + timestring + ">")
 
 
-def uid_to_username(uid: int) -> Text:
+def uid_to_username(uid: int) -> str:
     if uid in uid_to_username_cache:
         return uid_to_username_cache[uid]
 
@@ -342,10 +342,10 @@ def uid_to_username(uid: int) -> Text:
     return uid_to_username_cache[uid]
 
 
-def ps_line_to_process(ps_line: Text, now: datetime.datetime) -> PxProcess:
+def ps_line_to_process(ps_line: str, now: datetime.datetime) -> PxProcess:
     match = PS_LINE.match(ps_line)
     if not match:
-        raise Exception("Failed to match ps line <%r>" % ps_line)
+        raise Exception(f"Failed to match ps line <{ps_line:!r}>")
 
     process_builder = PxProcessBuilder()
     process_builder.pid = int(match.group(1))
@@ -464,7 +464,7 @@ def get_all() -> List[PxProcess]:
                 processes[process.pid] = process
 
             if ps.wait() != 0:
-                raise IOError("Exit code {} from {}".format(ps.returncode, command))
+                raise OSError(f"Exit code {ps.returncode} from {command}")
 
     resolve_links(processes, now)
     remove_process_and_descendants(processes, os.getpid())
@@ -484,7 +484,7 @@ def order_best_first(processes: Iterable[PxProcess]) -> List[PxProcess]:
     return ordered
 
 
-def seconds_to_str(seconds: float) -> Text:
+def seconds_to_str(seconds: float) -> str:
     if seconds < 60:
         seconds_s = str(seconds)
         decimal_index = seconds_s.rfind(".")
@@ -496,13 +496,13 @@ def seconds_to_str(seconds: float) -> Text:
     if seconds < 3600:
         minutes = int(seconds / 60)
         remaining_seconds = int(seconds - minutes * 60)
-        return "{}m{:02d}s".format(minutes, remaining_seconds)
+        return f"{minutes}m{remaining_seconds:02d}s"
 
     if seconds < 86400:
         hours = int(seconds / 3600)
         minutes = int((seconds - 3600 * hours) / 60)
-        return "{}h{:02d}m".format(hours, minutes)
+        return f"{hours}h{minutes:02d}m"
 
     days = int(seconds / 86400)
     hours = int((seconds - 86400 * days) / 3600)
-    return "{}d{:02d}h".format(days, hours)
+    return f"{days}d{hours:02d}h"

px/px_process_menu.py

@@ -207,10 +207,7 @@ def await_death(self, message):
             countdown_s = KILL_TIMEOUT_SECONDS - dt_s
             if countdown_s <= 0:
                 return
-            self.status = "{:.1f}s {}".format(
-                countdown_s,
-                message,
-            )
+            self.status = f"{countdown_s:.1f}s {message}"
             self.refresh_display()
 
             time.sleep(0.1)
@@ -231,15 +228,15 @@ def kill_process(
             )
             return
         self.await_death(
-            "Waiting for %s to shut down after SIGTERM" % self.process.command
+            f"Waiting for {self.process.command} to shut down after SIGTERM"
         )
         if not self.process.is_alive():
             return
 
         # Die!!
         assert signal_process(self.process, SIGKILL)
         self.await_death(
-            "Waiting for %s to shut down after kill -9" % self.process.command
+            f"Waiting for {self.process.command} to shut down after kill -9"
         )
         if not self.process.is_alive():
             return