mirror of
https://github.com/HorlogeSkynet/archey4
synced 2025-06-17 16:00:11 +02:00
[Output/entry] Refactors entry output slightly
Replaces callback with property
This commit is contained in:
@ -4,6 +4,7 @@ import json
|
||||
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 @@ class CPU(Entry):
|
||||
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 @@ class CPU(Entry):
|
||||
|
||||
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)
|
||||
|
@ -4,6 +4,7 @@ import logging
|
||||
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 @@ class Custom(Entry):
|
||||
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)
|
||||
|
@ -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,9 +227,10 @@ class Disk(Entry):
|
||||
|
||||
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.
|
||||
@ -236,8 +238,7 @@ class Disk(Entry):
|
||||
|
||||
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 @@ class Disk(Entry):
|
||||
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 @@ class Disk(Entry):
|
||||
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
|
||||
|
@ -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 @@ class Distro(Entry):
|
||||
|
||||
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")
|
||||
),
|
||||
)
|
||||
]
|
||||
|
@ -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 @@ class GPU(Entry):
|
||||
|
||||
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)
|
||||
|
@ -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 @@ class Kernel(Entry):
|
||||
|
||||
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 @@ class Kernel(Entry):
|
||||
else:
|
||||
text_output += f" ({self._default_strings.get('latest')})"
|
||||
|
||||
output.append(self.name, text_output)
|
||||
return [(self.name, text_output)]
|
||||
|
@ -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 @@ class LanIP(Entry):
|
||||
# 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 @@ class LanIP(Entry):
|
||||
else:
|
||||
text_output = self._default_strings.get("not_detected")
|
||||
|
||||
output.append(self.name, text_output)
|
||||
return [(self.name, text_output)]
|
||||
|
@ -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 @@ class LoadAverage(Entry):
|
||||
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
|
||||
]
|
||||
),
|
||||
)
|
||||
]
|
||||
|
@ -4,6 +4,7 @@ import os
|
||||
import platform
|
||||
import re
|
||||
from contextlib import suppress
|
||||
from functools import cached_property
|
||||
from subprocess import check_output
|
||||
from typing import Tuple
|
||||
|
||||
@ -152,14 +153,14 @@ class RAM(Entry):
|
||||
|
||||
return (mem_used / 1024), (mem_total / 1024)
|
||||
|
||||
def output(self, output) -> None:
|
||||
@cached_property
|
||||
def pretty_value(self) -> [(str, str)]:
|
||||
"""
|
||||
Adds the entry to `output` after pretty-formatting the RAM usage with color and units.
|
||||
Pretty-formats the RAM usage with color and units.
|
||||
"""
|
||||
if not self.value:
|
||||
# Fall back on the default behavior if no RAM usage could be detected.
|
||||
super().output(output)
|
||||
return
|
||||
return super().pretty_value
|
||||
|
||||
# DRY some constants
|
||||
used = self.value["used"]
|
||||
@ -173,6 +174,4 @@ class RAM(Entry):
|
||||
self.options.get("danger_use_percent", 66.7),
|
||||
)
|
||||
|
||||
output.append(
|
||||
self.name, f"{level_color}{int(used)} {unit}{Colors.CLEAR} / {int(total)} {unit}"
|
||||
)
|
||||
return [(self.name, f"{level_color}{int(used)} {unit}{Colors.CLEAR} / {int(total)} {unit}")]
|
||||
|
@ -5,6 +5,7 @@ import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
from functools import cached_property
|
||||
from glob import iglob
|
||||
from subprocess import DEVNULL, PIPE, CalledProcessError, check_output, run
|
||||
from typing import List, Optional
|
||||
@ -224,12 +225,12 @@ class Temperature(Entry):
|
||||
"""Simple Celsius to Fahrenheit conversion method"""
|
||||
return temp * (9 / 5) + 32
|
||||
|
||||
def output(self, output) -> None:
|
||||
"""Adds the entry to `output` after pretty-formatting with units."""
|
||||
@cached_property
|
||||
def pretty_value(self) -> [(str, str)]:
|
||||
"""Pretty-formats with units."""
|
||||
if not self.value:
|
||||
# Fall back on the default behavior if no temperatures were detected.
|
||||
super().output(output)
|
||||
return
|
||||
return super().pretty_value
|
||||
|
||||
# DRY some constants
|
||||
char_before_unit = self.value["char_before_unit"]
|
||||
@ -240,4 +241,4 @@ class Temperature(Entry):
|
||||
if len(self._temps) > 1:
|
||||
entry_text += f" (Max. {self.value['max_temperature']}{char_before_unit}{unit})"
|
||||
|
||||
output.append(self.name, entry_text)
|
||||
return [(self.name, entry_text)]
|
||||
|
@ -2,6 +2,7 @@
|
||||
|
||||
import os
|
||||
import re
|
||||
from functools import cached_property
|
||||
from typing import Optional
|
||||
|
||||
from archey.colors import Colors, Style
|
||||
@ -119,10 +120,11 @@ class Terminal(Entry):
|
||||
# Note : It _might_ be `None` in very specific environments.
|
||||
return env_term
|
||||
|
||||
def output(self, output) -> None:
|
||||
"""Adds the entry to `output` after pretty-formatting with colors palette"""
|
||||
@cached_property
|
||||
def pretty_value(self) -> [(str, str)]:
|
||||
"""Pretty-formats with colors palette"""
|
||||
text_output = self.value or self._default_strings.get("not_detected")
|
||||
if Style.should_color_output():
|
||||
text_output += " " + self._get_colors_palette()
|
||||
|
||||
output.append(self.name, text_output)
|
||||
return [(self.name, text_output)]
|
||||
|
@ -4,6 +4,7 @@ import re
|
||||
import time
|
||||
from contextlib import suppress
|
||||
from datetime import timedelta
|
||||
from functools import cached_property
|
||||
from subprocess import PIPE, run
|
||||
|
||||
from archey.entry import Entry
|
||||
@ -145,8 +146,9 @@ class Uptime(Entry):
|
||||
seconds=int(uptime_args.get("seconds") or 0),
|
||||
)
|
||||
|
||||
def output(self, output) -> None:
|
||||
"""Adds the entry to `output` after pretty-formatting the uptime to a string."""
|
||||
@cached_property
|
||||
def pretty_value(self) -> [(str, str)]:
|
||||
"""Pretty-formats the uptime to a string."""
|
||||
days = self.value["days"]
|
||||
hours = self.value["hours"]
|
||||
minutes = self.value["minutes"]
|
||||
@ -178,4 +180,4 @@ class Uptime(Entry):
|
||||
elif not days and not hours:
|
||||
uptime = "< 1 minute"
|
||||
|
||||
output.append(self.name, uptime)
|
||||
return [(self.name, uptime)]
|
||||
|
@ -1,5 +1,6 @@
|
||||
"""Public IP address detection class"""
|
||||
|
||||
from functools import cached_property
|
||||
from socket import timeout as SocketTimeoutError
|
||||
from subprocess import DEVNULL, CalledProcessError, TimeoutExpired, check_output
|
||||
from typing import Optional
|
||||
@ -96,17 +97,15 @@ class WanIP(Entry):
|
||||
except (URLError, SocketTimeoutError):
|
||||
return None
|
||||
|
||||
def output(self, output) -> None:
|
||||
"""Adds the entry to `output` after pretty-formatting our list of IP addresses."""
|
||||
@cached_property
|
||||
def pretty_value(self) -> [(str, str)]:
|
||||
"""Pretty-formats our list of IP addresses."""
|
||||
# If we found IP addresses, join them together nicely.
|
||||
# If not, fall-back on the "No address" string.
|
||||
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)
|
||||
|
||||
@ -115,4 +114,4 @@ class WanIP(Entry):
|
||||
else:
|
||||
text_output = self._default_strings.get("not_detected")
|
||||
|
||||
output.append(self.name, text_output)
|
||||
return [(self.name, text_output)]
|
||||
|
@ -3,6 +3,7 @@
|
||||
import logging
|
||||
from abc import ABC as AbstractBaseClass
|
||||
from abc import abstractmethod
|
||||
from functools import cached_property
|
||||
from typing import Optional
|
||||
|
||||
from archey.configuration import Configuration
|
||||
@ -36,12 +37,13 @@ class Entry(AbstractBaseClass):
|
||||
# Provision a logger for each entry.
|
||||
self._logger = logging.getLogger(self.__module__)
|
||||
|
||||
def output(self, output) -> None:
|
||||
"""Output the results to output. Can be overridden by subclasses."""
|
||||
@cached_property
|
||||
def pretty_value(self) -> [(str, str)]:
|
||||
"""Provide a "pretty" value. Can be overridden by subclasses."""
|
||||
if self.value:
|
||||
# Let's assume we can just use `__str__` on the object in value,
|
||||
# Let's assume we can just use `__str__` on the object in _value,
|
||||
# and create a single-line output with it.
|
||||
output.append(self.name, str(self.value))
|
||||
return [(self.name, str(self.value))]
|
||||
else:
|
||||
# If the value is "falsy" leave a generic "Not detected" message for this entry.
|
||||
output.append(self.name, self._default_strings.get("not_detected"))
|
||||
return [(self.name, self._default_strings.get("not_detected"))]
|
||||
|
@ -72,10 +72,6 @@ class Output:
|
||||
"""Append an entry to the list of entries to output"""
|
||||
self._entries.append(module)
|
||||
|
||||
def append(self, key: str, value) -> None:
|
||||
"""Append a pre-formatted entry to the final output content"""
|
||||
self._results.append(f"{self._entries_color}{key}:{Colors.CLEAR} {value}")
|
||||
|
||||
def output(self) -> None:
|
||||
"""
|
||||
Main `Output`'s `output` method.
|
||||
@ -85,9 +81,12 @@ class Output:
|
||||
if self._format_to_json:
|
||||
self._output_json()
|
||||
else:
|
||||
# Iterate through the entries and run their output method to add their content.
|
||||
# Iterate through the entries and get their content.
|
||||
for entry in self._entries:
|
||||
entry.output(self)
|
||||
for entry_line in entry.pretty_value:
|
||||
self._results.append(
|
||||
f"{self._entries_color}{entry_line[0]}:{Colors.CLEAR} {entry_line[1]}"
|
||||
)
|
||||
self._output_text()
|
||||
|
||||
def _output_json(self) -> None:
|
||||
|
Reference in New Issue
Block a user