[WIP] Output concurrency

archey/__main__.py

@@ -10,7 +10,7 @@
 import logging
 import os
 import sys
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from contextlib import ExitStack
 from enum import Enum
 from typing import Callable, Optional
@@ -124,6 +124,7 @@ def args_parsing() -> argparse.Namespace:
  return parser.parse_args()
 
 
+# todo: refactor out into functions/methods (too many branches)
 def main():
  """Simple entry point"""
  args = args_parsing()
@@ -152,13 +153,17 @@ def main():
  format_to_json=args.json,
  )
 
+ # Begin with some output as soon as possible
+ output.begin_output()
+
  # We will map this function onto our enabled entries to instantiate them.
- def _entry_instantiator(entry: dict) -> Optional[Entry]:
+ def _entry_instantiator(entry: dict, index: int) -> Optional[Entry]:
  # Based on **required** `type` field, instantiate the corresponding `Entry` object.
  try:
  return Entries[entry.pop("type")].value(
  name=entry.pop("name", None), # `name` is fully-optional.
  options=entry, # Remaining fields should be propagated as options.
+ index=index, # Provide entry with its position index.
  )
  except KeyError as key_error:
  logging.warning("One entry (misses or) uses an invalid `type` field (%s).", key_error)
@@ -182,11 +187,25 @@ def _entry_instantiator(entry: dict) -> Optional[Entry]:
  )
  mapper = executor.map
 
- for entry_instance in mapper(_entry_instantiator, available_entries):
- if entry_instance:
- output.add_entry(entry_instance)
+ # todo: solution fails if parallel_loading false because no executor
+ if configuration.get("output_concurrency"):
+ tasks = []
+ for index, entry in enumerate(available_entries):
+ future = executor.submit(_entry_instantiator, entry, index)
+ output.add_entry_concurrent(future)
+ tasks.append(future)
+
+ for future in as_completed(tasks):
+ output.entry_future_done_callback(future)
+
+ else:
+ for entry_instance in mapper(
+ _entry_instantiator, available_entries, range(len(available_entries))
+ ):
+ if entry_instance:
+ output.add_entry(entry_instance)
 
- output.output()
+ output.finish_output()
 
  # Has the screenshot flag been specified ?
  if args.screenshot is not None:

archey/colors.py

@@ -103,3 +103,45 @@ def __init__(self, bright: int, value: int):
  raise ValueError("Supplied color is outside the allowed range.")
  # `ESC[38;5` selects 8-bit foreground colour
  self.value = (bright, 38, 5, value)
+
+
+class TerminalMovements(Enum):
+ """
+ Enumeration of ANSI Terminal movement control sequences
+ """
+
+ UP = "A"
+ DOWN = "B"
+ FORWARD = "C"
+ BACK = "D"
+ NEXT_LINE = "E"
+ PREV_LINE = "F"
+ TO_COLUMN = "G"
+ TO_POSITION = "H"
+
+ ERASE_LINE = "K"
+
+ def move(self, *amounts):
+ """
+ Returns an ANSI escape code corresponding to the movement given.
+ Use `TerminalMovements` for the direction, and specify amounts 1-indexed, row before column.
+ """
+ display_attrs = ";".join(map(str, (*amounts, self.value)))
+ return f"\x1b[{display_attrs}"
+
+
+class CursorPosition(Style):
+ """
+ ANSI Terminal cursor positioning using escape sequences.
+ See <https://en.wikipedia.org/wiki/ANSI_escape_code#CSI_(Control_Sequence_Introducer)_sequences>
+ """
+
+ @staticmethod
+ def move(movement: TerminalMovements, *amounts):
+ """
+ Returns an ANSI escape code corresponding to the movement given.
+ Use `TerminalMovements` for the direction, and specify amounts 1-indexed, row before column.
+ """
+ # return Style.escape_code_from_attrs(";".join(map(str, (*amounts, movement.value))))
+ display_attrs = ";".join(map(str, (*amounts, movement.value)))
+ return f"\x1b[{display_attrs}"

archey/entries/cpu.py

@@ -4,6 +4,7 @@
 import platform
 import re
 from subprocess import DEVNULL, CalledProcessError, check_output
+from functools import cached_property
 from typing import Dict, List
 
 from archey.distributions import Distributions
@@ -184,12 +185,12 @@ def _parse_sysctl_cpu_model() -> List[Dict[str, int]]:
  model_name, nb_cores = sysctl_output.splitlines()
  return [{model_name: int(nb_cores)}]
 
- def output(self, output) -> None:
- """Writes CPUs to `output` based on preferences"""
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
+ """Provides CPU pretty value based on preferences"""
  # No CPU could be detected.
  if not self.value:
- output.append(self.name, self._default_strings.get("not_detected"))
- return
+ return [(self.name, self._default_strings.get("not_detected"))]
 
  entries = []
  for cpus in self.value:
@@ -201,8 +202,7 @@ def output(self, output) -> None:
 
  if self.options.get("one_line"):
  # One-line output is enabled : Join the results !
- output.append(self.name, ", ".join(entries))
+ return [(self.name, ", ".join(entries))]
  else:
  # One-line output has been disabled, add one entry per item.
- for entry in entries:
- output.append(self.name, entry)
+ return map(lambda entry: (self.name, entry), entries)

archey/entries/custom.py

@@ -4,6 +4,7 @@
 import os
 import stat
 from contextlib import suppress
+from functools import cached_property
 from subprocess import DEVNULL, PIPE, CalledProcessError, run
 from typing import List, Union
 
@@ -61,14 +62,13 @@ def __init__(self, *args, **kwargs):
  if log_stderr and proc.stderr:
  self._logger.warning("%s", proc.stderr.rstrip())
 
- def output(self, output) -> None:
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
  if not self.value:
- output.append(self.name, self._default_strings.get("not_detected"))
- return
+ return [(self.name, self._default_strings.get("not_detected"))]
 
  # Join the results only if `one_line` option is enabled.
  if self.options.get("one_line", True):
- output.append(self.name, ", ".join(self.value))
+ return [(self.name, ", ".join(self.value))]
  else:
- for element in self.value:
- output.append(self.name, element)
+ return map(lambda element: (self.name, element), self.value)

archey/entries/disk.py

@@ -3,6 +3,7 @@
 import platform
 import plistlib
 import re
+from functools import cached_property
 from subprocess import DEVNULL, PIPE, check_output, run
 from typing import Dict, Iterable, List
 
@@ -226,18 +227,18 @@ def _blocks_to_human_readable(blocks: float, suffix: str = "B") -> str:
 
  return f"{blocks:02.1f} {unit}{suffix}"
 
- def output(self, output) -> None:
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
  """
- Adds the entry to `output` after formatting with color and units.
+ Pretty-formats the entry with color and units.
  Follows the user configuration supplied for formatting.
  """
  # Fetch our `filesystems` object locally so we can modify it safely.
  filesystems = self.value
 
  if not filesystems:
  # We didn't find any disk, fall back to the default entry behavior.
- super().output(output)
- return
+ return super().pretty_value
 
  # DRY configuration object for the output.
  disk_labels = self.options.get("disk_labels")
@@ -270,6 +271,8 @@ def output(self, output) -> None:
  name += " "
  name += "({disk_label})"
 
+ entry_lines = []
+
  # We will only run this loop a single time for combined entries.
  for mount_point, filesystem_data in filesystems.items():
  # Select the corresponding level color based on disk percentage usage.
@@ -292,4 +295,6 @@ def output(self, output) -> None:
  self._blocks_to_human_readable(filesystem_data["total_blocks"]),
  )
 
- output.append(name.format(disk_label=disk_label), pretty_filesystem_value)
+ entry_lines.append((name.format(disk_label=disk_label), pretty_filesystem_value))
+
+ return entry_lines

archey/entries/distro.py

@@ -1,6 +1,7 @@
 """Distribution and architecture detection class"""
 
 import platform
+from functools import cached_property
 from subprocess import check_output
 from typing import Optional
 
@@ -43,10 +44,13 @@ def _fetch_darwin_release() -> Optional[str]:
 
  return f"Darwin {platform.release()}"
 
- def output(self, output) -> None:
- output.append(
- self.name,
- f"{{}} {self.value['arch']}".format(
- self.value["name"] or self._default_strings.get("not_detected")
- ),
- )
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
+ return [
+ (
+ self.name,
+ f"{{}} {self.value['arch']}".format(
+ self.value["name"] or self._default_strings.get("not_detected")
+ ),
+ )
+ ]

archey/entries/gpu.py

@@ -2,6 +2,7 @@
 
 import platform
 import re
+from functools import cached_property
 from subprocess import DEVNULL, CalledProcessError, check_output
 from typing import List
 
@@ -80,16 +81,15 @@ def _parse_pciconf_output() -> List[str]:
 
  return gpus_list
 
- def output(self, output) -> None:
- """Writes GPUs to `output` based on preferences"""
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
+ """Pretty-formats GPUs based on preferences"""
  # No GPU could be detected.
  if not self.value:
- output.append(self.name, self._default_strings.get("not_detected"))
- return
+ return [(self.name, self._default_strings.get("not_detected"))]
 
  # Join the results only if `one_line` option is enabled.
  if self.options.get("one_line"):
- output.append(self.name, ", ".join(self.value))
+ return [(self.name, ", ".join(self.value))]
  else:
- for gpu_device in self.value:
- output.append(self.name, gpu_device)
+ return map(lambda gpu_device: (self.name, gpu_device), self.value)

archey/entries/kernel.py

@@ -2,6 +2,7 @@
 
 import json
 import platform
+from functools import cached_property
 from socket import timeout as SocketTimeoutError
 from typing import Optional
 from urllib.error import URLError
@@ -56,7 +57,8 @@ def _fetch_latest_linux_release() -> Optional[str]:
 
  return kernel_releases.get("latest_stable", {}).get("version")
 
- def output(self, output) -> None:
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
  """Display running kernel and latest kernel if possible"""
  text_output = " ".join((self.value["name"], self.value["release"]))
 
@@ -66,4 +68,4 @@ def output(self, output) -> None:
  else:
  text_output += f" ({self._default_strings.get('latest')})"
 
- output.append(self.name, text_output)
+ return [(self.name, text_output)]

archey/entries/lan_ip.py

@@ -1,6 +1,7 @@
 """Local IP addresses detection class"""
 
 import ipaddress
+from functools import cached_property
 from itertools import islice
 from typing import Iterator
 
@@ -74,17 +75,15 @@ def _lan_ip_addresses_generator(
  # Finally, yield the address compressed representation.
  yield ip_addr.compressed
 
- def output(self, output) -> None:
- """Adds the entry to `output` after pretty-formatting the IP address list."""
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
+ """Pretty-formats the IP address list."""
  # If we found IP addresses, join them together nicely.
  # If not, fall back on default strings according to `netifaces` availability.
  if self.value:
  if not self.options.get("one_line", True):
  # One-line output has been disabled, add one IP address per item.
- for ip_address in self.value:
- output.append(self.name, ip_address)
-
- return
+ return map(lambda ip_address: (self.name, ip_address), self.value)
 
  text_output = ", ".join(self.value)
 
@@ -93,4 +92,4 @@ def output(self, output) -> None:
  else:
  text_output = self._default_strings.get("not_detected")
 
- output.append(self.name, text_output)
+ return [(self.name, text_output)]

archey/entries/load_average.py

@@ -2,6 +2,7 @@
 
 import os
 from contextlib import suppress
+from functools import cached_property
 
 from archey.colors import Colors
 from archey.entry import Entry
@@ -18,25 +19,27 @@ def __init__(self, *args, **kwargs):
  with suppress(AttributeError):
  self.value = os.getloadavg()
 
- def output(self, output) -> None:
+ @cached_property
+ def pretty_value(self) -> [(str, str)]:
  if not self.value:
  # Fall back on the default behavior if load average values could not be detected.
- super().output(output)
- return
+ return super().pretty_value
 
  # DRY constant thresholds.
  decimal_places = self.options.get("decimal_places", 2)
  warning_threshold = self.options.get("warning_threshold", 1.0)
  danger_threshold = self.options.get("danger_threshold", 2.0)
 
- output.append(
- self.name,
- " ".join(
- [
- str(Colors.get_level_color(load_avg, warning_threshold, danger_threshold))
- + str(round(load_avg, decimal_places))
- + str(Colors.CLEAR)
- for load_avg in self.value
- ]
- ),
- )
+ return [
+ (
+ self.name,
+ " ".join(
+ [
+ str(Colors.get_level_color(load_avg, warning_threshold, danger_threshold))
+ + str(round(load_avg, decimal_places))
+ + str(Colors.CLEAR)
+ for load_avg in self.value
+ ]
+ ),
+ )
+ ]