summaryrefslogtreecommitdiff
path: root/python/vyos
diff options
context:
space:
mode:
Diffstat (limited to 'python/vyos')
-rw-r--r--python/vyos/utils/convert.py145
-rw-r--r--python/vyos/utils/dict.py21
-rw-r--r--python/vyos/utils/io.py103
3 files changed, 269 insertions, 0 deletions
diff --git a/python/vyos/utils/convert.py b/python/vyos/utils/convert.py
new file mode 100644
index 000000000..975c67e0a
--- /dev/null
+++ b/python/vyos/utils/convert.py
@@ -0,0 +1,145 @@
+# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library. If not, see <http://www.gnu.org/licenses/>.
+
+def seconds_to_human(s, separator=""):
+ """ Converts number of seconds passed to a human-readable
+ interval such as 1w4d18h35m59s
+ """
+ s = int(s)
+
+ week = 60 * 60 * 24 * 7
+ day = 60 * 60 * 24
+ hour = 60 * 60
+
+ remainder = 0
+ result = ""
+
+ weeks = s // week
+ if weeks > 0:
+ result = "{0}w".format(weeks)
+ s = s % week
+
+ days = s // day
+ if days > 0:
+ result = "{0}{1}{2}d".format(result, separator, days)
+ s = s % day
+
+ hours = s // hour
+ if hours > 0:
+ result = "{0}{1}{2}h".format(result, separator, hours)
+ s = s % hour
+
+ minutes = s // 60
+ if minutes > 0:
+ result = "{0}{1}{2}m".format(result, separator, minutes)
+ s = s % 60
+
+ seconds = s
+ if seconds > 0:
+ result = "{0}{1}{2}s".format(result, separator, seconds)
+
+ return result
+
+def bytes_to_human(bytes, initial_exponent=0, precision=2):
+ """ Converts a value in bytes to a human-readable size string like 640 KB
+
+ The initial_exponent parameter is the exponent of 2,
+ e.g. 10 (1024) for kilobytes, 20 (1024 * 1024) for megabytes.
+ """
+
+ if bytes == 0:
+ return "0 B"
+
+ from math import log2
+
+ bytes = bytes * (2**initial_exponent)
+
+ # log2 is a float, while range checking requires an int
+ exponent = int(log2(bytes))
+
+ if exponent < 10:
+ value = bytes
+ suffix = "B"
+ elif exponent in range(10, 20):
+ value = bytes / 1024
+ suffix = "KB"
+ elif exponent in range(20, 30):
+ value = bytes / 1024**2
+ suffix = "MB"
+ elif exponent in range(30, 40):
+ value = bytes / 1024**3
+ suffix = "GB"
+ else:
+ value = bytes / 1024**4
+ suffix = "TB"
+ # Add a new case when the first machine with petabyte RAM
+ # hits the market.
+
+ size_string = "{0:.{1}f} {2}".format(value, precision, suffix)
+ return size_string
+
+def human_to_bytes(value):
+ """ Converts a data amount with a unit suffix to bytes, like 2K to 2048 """
+
+ from re import match as re_match
+
+ res = re_match(r'^\s*(\d+(?:\.\d+)?)\s*([a-zA-Z]+)\s*$', value)
+
+ if not res:
+ raise ValueError(f"'{value}' is not a valid data amount")
+ else:
+ amount = float(res.group(1))
+ unit = res.group(2).lower()
+
+ if unit == 'b':
+ res = amount
+ elif (unit == 'k') or (unit == 'kb'):
+ res = amount * 1024
+ elif (unit == 'm') or (unit == 'mb'):
+ res = amount * 1024**2
+ elif (unit == 'g') or (unit == 'gb'):
+ res = amount * 1024**3
+ elif (unit == 't') or (unit == 'tb'):
+ res = amount * 1024**4
+ else:
+ raise ValueError(f"Unsupported data unit '{unit}'")
+
+ # There cannot be fractional bytes, so we convert them to integer.
+ # However, truncating causes problems with conversion back to human unit,
+ # so we round instead -- that seems to work well enough.
+ return round(res)
+
+def mac_to_eui64(mac, prefix=None):
+ """
+ Convert a MAC address to a EUI64 address or, with prefix provided, a full
+ IPv6 address.
+ Thankfully copied from https://gist.github.com/wido/f5e32576bb57b5cc6f934e177a37a0d3
+ """
+ import re
+ from ipaddress import ip_network
+ # http://tools.ietf.org/html/rfc4291#section-2.5.1
+ eui64 = re.sub(r'[.:-]', '', mac).lower()
+ eui64 = eui64[0:6] + 'fffe' + eui64[6:]
+ eui64 = hex(int(eui64[0:2], 16) ^ 2)[2:].zfill(2) + eui64[2:]
+
+ if prefix is None:
+ return ':'.join(re.findall(r'.{4}', eui64))
+ else:
+ try:
+ net = ip_network(prefix, strict=False)
+ euil = int('0x{0}'.format(eui64), 16)
+ return str(net[euil])
+ except: # pylint: disable=bare-except
+ return
diff --git a/python/vyos/utils/dict.py b/python/vyos/utils/dict.py
index 66b40d92b..4afc9f54e 100644
--- a/python/vyos/utils/dict.py
+++ b/python/vyos/utils/dict.py
@@ -233,3 +233,24 @@ def dict_to_list(d, save_key_to=None):
collect.append(item)
return collect
+
+def check_mutually_exclusive_options(d, keys, required=False):
+ """ Checks if a dict has at most one or only one of
+ mutually exclusive keys.
+ """
+ present_keys = []
+
+ for k in d:
+ if k in keys:
+ present_keys.append(k)
+
+ # Un-mangle the keys to make them match CLI option syntax
+ from re import sub
+ orig_keys = list(map(lambda s: sub(r'_', '-', s), keys))
+ orig_present_keys = list(map(lambda s: sub(r'_', '-', s), present_keys))
+
+ if len(present_keys) > 1:
+ raise ValueError(f"Options {orig_keys} are mutually-exclusive but more than one of them is present: {orig_present_keys}")
+
+ if required and (len(present_keys) < 1):
+ raise ValueError(f"At least one of the following options is required: {orig_present_keys}")
diff --git a/python/vyos/utils/io.py b/python/vyos/utils/io.py
new file mode 100644
index 000000000..843494855
--- /dev/null
+++ b/python/vyos/utils/io.py
@@ -0,0 +1,103 @@
+# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library. If not, see <http://www.gnu.org/licenses/>.
+
+def print_error(str='', end='\n'):
+ """
+ Print `str` to stderr, terminated with `end`.
+ Used for warnings and out-of-band messages to avoid mangling precious
+ stdout output.
+ """
+ import sys
+ sys.stderr.write(str)
+ sys.stderr.write(end)
+ sys.stderr.flush()
+
+def make_progressbar():
+ """
+ Make a procedure that takes two arguments `done` and `total` and prints a
+ progressbar based on the ratio thereof, whose length is determined by the
+ width of the terminal.
+ """
+ import shutil, math
+ col, _ = shutil.get_terminal_size()
+ col = max(col - 15, 20)
+ def print_progressbar(done, total):
+ if done <= total:
+ increment = total / col
+ length = math.ceil(done / increment)
+ percentage = str(math.ceil(100 * done / total)).rjust(3)
+ print_error(f'[{length * "#"}{(col - length) * "_"}] {percentage}%', '\r')
+ # Print a newline so that the subsequent prints don't overwrite the full bar.
+ if done == total:
+ print_error()
+ return print_progressbar
+
+def make_incremental_progressbar(increment: float):
+ """
+ Make a generator that displays a progressbar that grows monotonically with
+ every iteration.
+ First call displays it at 0% and every subsequent iteration displays it
+ at `increment` increments where 0.0 < `increment` < 1.0.
+ Intended for FTP and HTTP transfers with stateless callbacks.
+ """
+ print_progressbar = make_progressbar()
+ total = 0.0
+ while total < 1.0:
+ print_progressbar(total, 1.0)
+ yield
+ total += increment
+ print_progressbar(1, 1)
+ # Ignore further calls.
+ while True:
+ yield
+
+def ask_input(question, default='', numeric_only=False, valid_responses=[]):
+ question_out = question
+ if default:
+ question_out += f' (Default: {default})'
+ response = ''
+ while True:
+ response = input(question_out + ' ').strip()
+ if not response and default:
+ return default
+ if numeric_only:
+ if not response.isnumeric():
+ print("Invalid value, try again.")
+ continue
+ response = int(response)
+ if valid_responses and response not in valid_responses:
+ print("Invalid value, try again.")
+ continue
+ break
+ return response
+
+def ask_yes_no(question, default=False) -> bool:
+ """Ask a yes/no question via input() and return their answer."""
+ from sys import stdout
+ default_msg = "[Y/n]" if default else "[y/N]"
+ while True:
+ try:
+ stdout.write("%s %s " % (question, default_msg))
+ c = input().lower()
+ if c == '':
+ return default
+ elif c in ("y", "ye", "yes"):
+ return True
+ elif c in ("n", "no"):
+ return False
+ else:
+ stdout.write("Please respond with yes/y or no/n\n")
+ except EOFError:
+ stdout.write("\nPlease respond with yes/y or no/n\n")