Compare commits

...

29 commits

Author SHA1 Message Date
Martin Sekera
8f505849af over-3.0
Text output API was brought in line with ACE and similar libraries: there are
different functions for each log level, all with prinf semantics.

Unit is now fully usable with base-2 (MiB) and base-10 (MB) systems, as well
as arbitrary-base logarithms (dBm).

ProgressBar has been updated to expose more Unit configuration.
2021-08-24 09:23:29 +02:00
Martin Sekera
852cce04b6 over.app.Option: add new source type (preset) 2021-04-07 23:05:08 +02:00
Martin Sekera
03ab783437 fix indent in over.file.filelist 2021-03-24 16:37:37 +01:00
Martin Sekera
f3fdf22e53 add over.file.filelist function 2021-03-24 16:26:25 +01:00
Martin Sekera
10311ffd46 silence warnings in cmd.Command 2021-02-19 01:53:47 +01:00
Martin Sekera
2645383a6f fix callback.path ignoring the 'exists' attr on writable paths 2021-02-16 17:58:09 +01:00
Martin Sekera
3f6b7cd24e add Main.info(), Main.done(), Main.warn(), Main.fail() log aliases 2021-02-16 17:56:59 +01:00
Martin Sekera
396ae4e49d over.app: add .dirs with the app's xdg dirs 2021-02-15 00:29:35 +01:00
Martin Sekera
b7c3ca3e02 fix over.cmd.Command.dump not quoting empty strings 2020-09-15 01:50:32 +02:00
Martin Sekera
b083fbc384 Merge branch 'master' of git.decade.cz:overwatch/over 2020-03-09 07:43:28 +01:00
Martin Sekera
ca06e2438a minor hexdump UI fix 2020-03-09 07:42:49 +01:00
Martin Sekera
0f1d72149e fix hexdump offset being off by one line 2020-03-07 18:05:22 +01:00
Martin
193302708c over.cfg.Config - add .items() and .values() 2019-08-07 11:53:39 +02:00
Martin
43ba8ee6a3 add over.cfg.Config - a simple read-only API for JSON configs 2019-08-06 17:38:28 +02:00
Martin
88be91445f Merge branch 'master' of decade.cz:overwatch/over 2019-05-29 13:50:43 +02:00
Martin Sekera
7634b3e35d add over.cmd.Command.__str__ 2019-04-23 01:51:51 +02:00
Martin Sekera
908857a317 changed the over.callback.path API some more 2019-04-18 02:18:15 +02:00
Martin Sekera
cf9fda165f changed the over.callback.path API a bit 2019-04-18 02:08:20 +02:00
Martin Sekera
1d5090b33e generalized over.callbacks.directory -> over.callbacks.path 2019-04-18 01:57:58 +02:00
Martin Sekera
430ed6e34d hexdump: skip lines if the content is the same 2019-03-31 00:13:55 +01:00
Martin
e8ce658015 Merge branch 'master' of decade.cz:overwatch/over 2019-02-06 09:26:34 +01:00
Martin Sekera
f31e2137ac add over.app.Main.setup arg reset 2018-12-12 22:40:28 +01:00
Martin
5ed3e16d65 add over.callback.integers 2017-11-23 16:01:21 +01:00
Martin
cc7d97c811 chasing own tail 2017-10-16 10:49:24 +02:00
Martinez
cfc2197943 more minor doc fixes 2017-10-15 10:17:01 +02:00
Martinez
37f36ef301 minor string fix 2017-10-15 10:00:39 +02:00
Martin
1ed1087a1a fix missing over.types.ndict.keys 2017-10-13 18:23:08 +02:00
Martin
f8b9f983ad improved over.types.ndict.__repr__ 2017-10-13 14:28:15 +02:00
Martinez
c92c0fb6e9 update .gitignore 2017-10-10 00:44:03 +02:00
13 changed files with 401 additions and 177 deletions

4
.gitignore vendored
View file

@ -1,3 +1,3 @@
*.so
*.pyc
__pycache__
/MANIFEST
/dist

View file

@ -6,6 +6,7 @@ import sys
from . import app
from . import aux
from . import callback
from . import cfg
from . import cmd
from . import docs
from . import file

View file

@ -107,6 +107,7 @@ class Option_sources(enum.Enum):
default = 1
config_file = 2
command_line = 3
preset = 4
class Option:
def __init__(self, name, description, callback, default=Option_sources.none, count=0, overwrite=True, abbr=None, in_cfg_file=True, in_help=True, is_boolean=None):
@ -219,7 +220,7 @@ def expand_group(token, options):
options_by_abbr = {option.abbr: option for option in options.values() if option.abbr}
group_positive = token[0] is "+"
group_positive = token[0] == "+"
for abbr in token[1:]:
try:
@ -304,7 +305,7 @@ class ConfigFile:
self.options = options
self.path = path
self.ignore_unknown = ignore_unknown
self.print = text.Output("over.app.ConfigFile")
self.log = text.Log("over.app.ConfigFile")
self.seen_hashes = set()
def read_config(self):
@ -355,13 +356,13 @@ class ConfigFile:
config_dir = os.path.dirname(self.path)
if config_dir.strip() and not os.path.exists(config_dir):
os.mkdir(config_dir)
self.print("created config directory <c>%s<.>" %(config_dir))
self.log.info("created config directory <c>%s<.>", config_dir)
# if the file doesn't exist, create it with a boilerplate header
if not os.path.exists(self.path):
with open(self.path, "w") as f:
f.write(header_template %header_args)
self.print("created empty config file <c>%s<.>" %(self.path))
self.log.info("created empty config file <c>%s<.>", self.path)
# add new or otherwise missing options
updated = set()
@ -369,7 +370,7 @@ class ConfigFile:
with open(self.path, "a") as f:
for option in self.options.values():
if option.hash not in self.seen_hashes and option.in_cfg_file:
self.print("adding <W>--<G>%s<.> to config file" %(option.name))
self.log.info("adding <W>--<G>%s<.> to config file", option.name)
f.write(docs.config_file_item %(
option.hash,
@ -414,7 +415,7 @@ class Main:
self.name = name
self.version = version
self.license = license
self.print = text.Output(name)
self.log = text.Log(name)
self.options = OrderedDict()
self.options_by_abbr = OrderedDict()
self.docs = OrderedDict()
@ -425,6 +426,7 @@ class Main:
self.last_command_line = None
self.using_alternate_config = False
self.uncontained_exception_callbacks = [] # (function, args)
self.dirs = get_xdg_paths(name)
for feature_name in self.default_features:
if feature_name in features:
@ -509,7 +511,7 @@ class Main:
for target in self.targets:
tokens.append(target)
self.print(" ".join(tokens), format="<t>")
self.log.write(" ".join(tokens), format="<t>")
def activate_debug(self):
raise NotImplementedError("debug is not yet here")
@ -538,23 +540,23 @@ class Main:
@while displaying help
"""
print = text.Output("help", format="<t>", end="\n")
print = text.Log("help", format="<t>", end="\n").write
# App name and version
print("[<W>Application<.>]")
print(" <W>%s<.>-<c>%s<.> licensed under the <W>%s<.>" %(self.name, self.version, self.license))
print(" <W>%s<.>-<c>%s<.> licensed under <W>%s<.>", self.name, self.version, self.license)
print(" using over-%s" %(version.str))
# Main features
print("")
print("[<W>over.app.Main features<.>]")
for feature_name in self.features:
print(" %s: <B>%s<.> (%s)" %(" <G>ON<.>" if self.features[feature_name] else "<R>OFF<.>", feature_name, self.default_features[feature_name][1]))
print(" %s: <B>%s<.> (%s)", " <G>ON<.>" if self.features[feature_name] else "<R>OFF<.>", feature_name, self.default_features[feature_name][1])
# App docs
print("")
for chapter, paragraphs in (alternate_docs or self.docs).items():
print("[<W>%s<.>]" %(chapter))
print("[<W>%s<.>]", chapter)
for paragraph in paragraphs:
print(paragraph, format=" <i><t>")
@ -601,10 +603,10 @@ class Main:
else:
colored_current_value = '<M>' + colored_current_value + '<.>'
print(" Current value (%s): %s" %(
print(" Current value (%s): %s",
serialize_source(option.source),
colored_current_value
))
)
# some misc flags
if not option.in_cfg_file:
@ -618,7 +620,7 @@ class Main:
# Current targets, if any
if self.targets:
print("[<W>Targets<.>]")
print(" <M>%s<.>" %(cmd.format_invocation(self.targets)))
print(" <M>%s<.>", cmd.format_invocation(self.targets))
self.exit()
@ -726,11 +728,15 @@ class Main:
for action in scheduled_actions:
action()
def setup(self, command_line=None, force_config=None):
def setup(self, command_line=None, force_config=None, reset=False):
"""
@while determining application configuration
"""
if reset:
for option in self.options.values():
option.reset(True)
self.parse_config(force_config)
self.parse_command_line(command_line)
@ -743,7 +749,7 @@ class Main:
If things go south even here, prints the topmost traceback and gives up in a safe manner.
"""
self.print("<R>uncontained exception<.> <Y>%s<.> raised" %(exception_type.__name__), self.print.tl.fail)
self.log.fail("<R>uncontained exception<.> <Y>%s<.> raised", exception_type.__name__)
try:
tb_lines = ["", "---------------- Stack trace ----------------", "In program <W>%s<.>" %(self.invocation)]
@ -803,26 +809,26 @@ class Main:
else:
format = "%s - <i><t>" %((i - 3) * " ")
self.print(line, format=format, end="\n")
self.log.write(line, format=format, end="\n")
self.print("---------------------------------------------", format="<t>", end="\n")
self.log.write("---------------------------------------------", format="<t>", end="\n")
except:
self.print("<R>failed to contain exception<.>", self.print.tl.epic)
self.log.epic("<R>failed to contain exception<.>")
traceback.print_exc()
if self.uncontained_exception_callbacks:
self.print("executing containment callbacks", self.print.tl.exec)
self.log.start("executing containment callbacks")
l = len(self.uncontained_exception_callbacks)
for i, (cb, ctx) in enumerate(self.uncontained_exception_callbacks):
self.print("(%d/%d) %s" %(i+1, l, cb.__name__))
self.log.write("(%d/%d) %s", i+1, l, cb.__name__)
try:
cb(*ctx)
except Exception as e:
self.print("(%d/%d) <r>%s failed<.> (<R>%s<.>)" %(i+1, l, cb.__name__, e.__class__.__name__), self.print.tl.epic, end=":\n")
self.log.epic("(%d/%d) <r>%s failed<.> (<R>%s<.>)", i+1, l, cb.__name__, e.__class__.__name__, end=":\n")
exc_info = sys.exc_info()
traceback.print_exception(*exc_info)
del exc_info
self.print("containment callbacks executed", self.print.tl.done)
self.log.done("containment callbacks executed")

View file

@ -7,7 +7,7 @@ import traceback
from . import text
class DeprecationForwarder:
print = text.Output("over.aux.DeprecationForwarder", stream=sys.stderr)
log = text.Log("over.aux.DeprecationForwarder")
def __init__(self, target, old_name, new_name):
self._target = target
@ -16,6 +16,6 @@ class DeprecationForwarder:
def __getattr__(self, name):
caller = traceback.extract_stack()[-2]
self.print("%s is deprecated, please use %s instead (%s:%d)" %(self._old_name, self._new_name, caller[0], caller[1]), self.print.tl.warn)
self.log.warn("%s is deprecated, please use %s instead (%s:%d)", self._old_name, self._new_name, caller[0], caller[1])
return getattr(self._target, name)

View file

@ -53,27 +53,62 @@ def strings(*args):
return out
def directory(exists=False, writable=False, gio=False):
def path(exists=False, permissions=None, validators=None):
"""
Returns a directory callback that raises hell if:
- the supplied directory doesn't exist and `exists` is True
- isn't writable and `writable` is True
- isn't a valid Gio path and `gio` is True
Returns a path callback that takes a path (str) and verifies if it:
- exists iff `exists` is True
- its permissions match those in `permissions` ("rwx" or any subset, e.g. "r--", "-w-", dashes are optional)
- goes through each (function, message) pair in `validators`, in order
@while generating a callback
A validator is a function that takes a str argument and returns True or False.
If False is returned, the matching message is raised along with a RuntimeError.
The message may contain %s to be replaced by the supplied path. Example:
validators=[
(os.path.isdir, "%s is not a directory"),
(os.path.isabs, "%s is a directory, but is not an absolute path.")
]
@while generating a path callback
"""
if gio:
raise NotImplementedError("Gio support is not yet here")
def cb(arg):
path = os.path.abspath(os.path.expanduser(arg))
if not os.path.isdir(path) and exists:
raise FileNotFoundError("%s (%s) does not exist" %(arg, path))
if exists and not os.path.exists(path):
raise FileNotFoundError("%s does not exist" %(arg))
if writable and not os.access(path, os.W_OK | os.X_OK):
raise PermissionError("%s exists but is not writable" %(arg))
if permissions:
if "r" in permissions and not os.access(path, os.R_OK):
raise PermissionError("%s is not readable" %(arg))
if exists or os.path.exists(path):
if "w" in permissions and not os.access(path, os.W_OK):
raise PermissionError("%s is not writable" %(arg))
else:
if "w" in permissions and not os.access(os.path.dirname(path), os.W_OK):
raise PermissionError("%s cannot be created" %(arg))
if "x" in permissions and not os.access(path, os.X_OK):
raise PermissionError("%s is not executable" %(arg))
if validators:
if hasattr(validators, "__iter__"):
src = validators
else:
src = [validators]
for v in src:
if hasattr(v, "__iter__"):
fn, msg_tpl = v
else:
fn = v
msg_tpl = "%%s failed validation by '%s'" %(fn.__name__)
if not fn(path):
msg = msg_tpl %(arg) if "%s" in msg_tpl else msg_tpl
raise RuntimeError(msg)
return path
@ -99,3 +134,15 @@ def integer(arg):
return int(arg, 8)
return int(arg)
def integers(*args):
"""
@while converting arguments to ints
"""
out = []
for arg in args:
out.append(integer(arg))
return out

75
over/cfg.py Normal file
View file

@ -0,0 +1,75 @@
#! /usr/bin/env python3
# encoding: utf-8
# --------------------------------------------------
# Library imports
import json
import jsmin
# --------------------------------------------------
# Local imports
# --------------------------------------------------
class Config:
def __init__(self, source=None, readonly=False):
"""
Can be loaded from a JSON file (str path) or from a python dict.
"""
if type(source) == str:
with open(source) as f:
initial = json.loads(jsmin.jsmin(f.read()))
elif source is None:
initial = {}
else:
initial = source
assert type(initial) == dict
object.__setattr__(self, "raw", initial)
object.__setattr__(self, "readonly", readonly)
def keys(self):
return self.raw.keys()
def items(self):
return self.raw.items()
def values(self):
return self.raw.values()
def __getitem__(self, name):
return self.raw[name]
def __setitem__(self, name, value):
if self.readonly:
raise AttributeError("object is not writable")
self.raw[name] = value
def __setattr__(self, name, value):
if self.readonly:
raise AttributeError("object is not writable")
self.raw[name] = value
def __getattr__(self, name):
matches = [key for key in self.raw if key.replace("-", "_") == name]
if matches:
assert len(matches) == 1
value = self.raw[matches[0]]
if type(value) == dict:
return Config(value)
else:
return value
else:
raise KeyError(name)
def __contains__(self, name):
return name in self.raw
def __repr__(self):
return "Config(%s)" %(" ".join("%s=%s" %(k, "" if type(v) == dict else v) for k, v in self.raw.items()))

View file

@ -30,10 +30,13 @@ def char_in_str(chars, string):
# --------------------------------------------------
def format_invocation(args):
def escape_quote_invocation(args):
escaped = (arg.replace('"', '\\"') for arg in args)
return " ".join(('"%s"' %(a) if char_in_str(' $()[];\\"', a) else a) for a in escaped)
return [('"%s"' %(a) if char_in_str(' $()[];\\"|', a) else a) for a in escaped]
def format_invocation(args):
return " ".join(escape_quote_invocation(args))
# --------------------------------------------------
@ -61,6 +64,9 @@ class Command:
self.__dict__["sequence_original"] = list(sequence)
self.reset()
def __str__(self):
return " ".join(self.dump(pretty=True))
def __setattr__(self, name, value):
found = False
@ -94,7 +100,8 @@ class Command:
out.append(str(item))
if pretty:
return [('"%s"' %(a) if char_in_str(" $()[];\\", a) else a) for a in out]
return escape_quote_invocation(out)
# return [('"%s"' %(a) if (not a or char_in_str(" $()[];\\", a)) else a) for a in out]
else:
return out
@ -105,7 +112,7 @@ class Command:
stderr capture stderr instead of stdout (can't do both yet)
"""
self.__dict__["process"] = subprocess.Popen(self.dump(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1)
self.__dict__["process"] = subprocess.Popen(self.dump(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=16)
self.__dict__["fifo"] = queue.Queue()
self.__dict__["thread"] = threading.Thread(
target=capture_output,

View file

@ -13,7 +13,7 @@ over_docs["Options"] = [
"Boolean options can accept 0 words - then they become flags (switches) and are toggled directly by their names: <W>--<g>enabled<.> evaluates to <M>True<.>, <R>--no-<g>enabled<.> to <M>False<.>.",
"Other flags (options that accept 0 words) directly trigger callbacks. Those are usually referred to as actions. A good example is <W>--<g>help<.>.",
"Options can be referred to by their abbreviated, single character names, if defined by the application. Abbreviated names are interchangeable with full names. Boolean flags get shortened from <W>--<g>enabled<.> to <W>+<G>E<.>, and <R>--no-<g>enabled<.> to <R>-<G>E<.>.",
"Multiple abbreviated options can be grouped together, with one condition: all except the last (rightmost) option must be flags or actions. For example <R>-<G>Exp<.> <M>primary<.> is the same as <R>--no-<g>enabled<.> <R>--no-<g>execute<.> <W>--<g>pool<.> <M>primary<.> or <W>+<G>Nv<.> <M>drop<.> expands to <W>--<g>normalize<.> <W>--<g>video<.> <M>drop<.>. Notice how the entire group takes on the boolean value of the leading <W>+<.> or <R>-<.>.",
"Multiple abbreviated options can be grouped together, with one condition: all except the last (rightmost) option must be flags or actions. For example <R>-<G>Exp<.> <M>primary<.> is the same as <R>--no-<g>enabled<.> <R>--no-<g>execute<.> <W>--<g>pool<.> <M>primary<.>, or <W>+<G>Nv<.> <M>drop<.> expands to <W>--<g>normalize<.> <W>--<g>video<.> <M>drop<.>. Notice how the entire group takes on the boolean value of the leading <W>+<.> or <R>-<.>.",
'If an option is overwriting (<c>over<.>.<c>app<.>.<c>Option<.>.<y>overwrite<.> = <M>True<.>) the latest (rightmost) instance of the same option overwrites all previous ones. Otherwise, all instances are used. For example, <W>+<G>v<.> <M>drop<.> <W>+<G>v<.> <M>copy<.> <W>+<G>v<.> <M>x264<.> evaluates to <M>["x264"]<.> if the option is defined as overwriting, whereas if it was not it would evaluate to <M>["drop", "copy", "x264"]<.>.'
]
over_docs["Configuration sources"] = ["Each option can take its state either from (in order) the application default value, the config file, or the command line. Combining multiple sources (e.g. extending values in the config file with the command line) is not permitted - a new source always resets the option's state. It is possible for an option to have no value."]
@ -22,14 +22,14 @@ over_docs["Config File"] = ["If enabled by the application, a config file will b
colors = []
for code, (_, name) in ansi_colors.items():
c = "<C><<%s><.>: <%s>%s<.>" %(code, code, name)
c = "<<%s> = <%s>%s<.>" %(code, code, name)
if code == "k":
c += ' (this one says "black")'
colors.append(c)
over_docs["Available Colors"] = colors
over_docs["Available Colors"] = [", ".join(colors)]
config_file_header = """# Configuration file for %s-%s
# Generated by over-%s

View file

@ -54,3 +54,18 @@ def context_dir(path):
yield
finally:
os.chdir(previous)
# --------------------------------------------------
def filelist(d):
"""
Returns a filelist of d.
"""
files = []
for root, directories, filenames in os.walk(d):
for filename in filenames:
files.append(os.path.join(root, filename))
return files

View file

@ -110,16 +110,17 @@ def debugger():
# --------------------------------------------------
def hexdump(data, indent=0, offset=16, show_header=True, show_offsets=True, show_ascii=True, use_colors=True, initial_offset=0, output=sys.stdout):
def hexdump(data, indent=0, width=16, show_header=True, show_offsets=True, show_ascii=True, use_colors=True, initial_offset=0, output=sys.stdout, skip_same=True):
"""
Writes a hex dump of `data` to `output`.
The output text is indented with spaces and contains `offset` bytes per line.
The output text is indented with spaces and contains `width` bytes per line.
If `show_header` is True, a single line with byte numbers preceeds all output.
If `show_offsets` is True, each line is prefixed with the address of the first byte of that line.
If `show_ascii` is True, each line is suffixed with its ASCII representation. Unprintable characters
are replaced with a dot.
The `output` must implement a .write(str x) method. If `output` is None, the string is returned instead.
If `skip_same` is True, lines with identical content are abbreviated (omitted).
@while creating a hex dump
"""
@ -145,25 +146,22 @@ def hexdump(data, indent=0, offset=16, show_header=True, show_offsets=True, show
if show_offsets:
line.append("offset"[:offset_figures].ljust(offset_figures + 2))
for i in range(offset):
for i in range(width):
line.append("%2x" %(i))
if show_ascii:
line.append(" *{0}*".format("ASCII".center(offset, "-")))
line.append(" *{0}*".format("ASCII".center(width, "-")))
output_io.write(" ".join(line) + "\n")
skip_cnt = 0
last_hex_bytes = None
while data[ptr:]:
if indent:
output_io.write(" " * indent)
if show_offsets:
output_io.write(format_str %(initial_offset + ptr))
hex_bytes = []
ascii_bytes = []
for local_i, i in enumerate(range(ptr, ptr+offset)):
for local_i, i in enumerate(range(ptr, ptr+width)):
if i < len(data):
c = data[i]
hex_bytes.append("%02x" %(c))
@ -173,7 +171,30 @@ def hexdump(data, indent=0, offset=16, show_header=True, show_offsets=True, show
else:
ascii_bytes.append(".")
elif i == len(data):
hex_bytes.extend([" "] * (offset - local_i))
hex_bytes.extend([" "] * (width - local_i))
ptr += width
# the rest is just rendering
if skip_same:
if hex_bytes == last_hex_bytes:
skip_cnt += 1
else:
if skip_cnt:
output_io.write("%s > skipping %d identical lines, %d B\n" %(" "*offset_figures, skip_cnt, skip_cnt * width))
skip_cnt = 0
last_hex_bytes = hex_bytes
if skip_cnt:
continue
if indent:
output_io.write(" " * indent)
if show_offsets:
output_io.write(format_str %(initial_offset + ptr - width))
if use_colors: output_io.write(text.render("<W>"))
output_io.write(" ".join(hex_bytes))
@ -186,7 +207,8 @@ def hexdump(data, indent=0, offset=16, show_header=True, show_offsets=True, show
output_io.write("\n")
ptr += offset
if skip_cnt:
output_io.write("%s > skipping %d identical lines, %d B\n" %(" "*offset_figures, skip_cnt, skip_cnt * width))
if not output:
output_io.seek(0)

View file

@ -15,16 +15,16 @@ import tzlocal
# --------------------------------------------------
def lexical_join(words, oxford=False):
def lexical_join(words, oxford=True):
"""
Joins an iterable of words or sentence fragments into a lexical list:
>>> lexical_join(["this", "that", "one of them too"])
"this, that and one of them too"
>>> lexical_join(["this", "that", "one of them too"], oxford=True)
"this, that, and one of them too"
>>> lexical_join(["this", "that", "one of them too"], oxford=False)
"this, that and one of them too"
>>> lexical_join(["this", "that"])
"this and that"
@ -49,66 +49,80 @@ def lexical_join(words, oxford=False):
class Unit:
"""
A object that represents numbers and units in human-readable form.
TODO use significant digits instead of rigid order boundaries
TODO float superclass?
TODO base_2 prefixes (Ki, Mi, ...)
"""
_prefixes = (
_prefixes_base2 = (
("Yi", 80), ("Zi", 70), ("Ei", 60), ("Pi", 50), ("Ti", 40), ("Gi", 30), ("Mi", 20), ("Ki", 10),
("", 0)
)
_prefixes_base10 = (
("Y", 24), ("Z", 21), ("E", 18), ("P", 15), ("T", 12), ("G", 9), ("M", 6), ("k", 3),
("h", 2), ("D", 1), ("", 0), ("d", -1), ("c", -2), ("m", -3), ("μ", -6), ("n", -9),
("", 0), ("d", -1), ("c", -2), ("m", -3), ("µ", -6), ("n", -9),
("p", -12), ("f", -15), ("a", -18), ("z", -21), ("y", -24)
)
def __init__(self,
value, unit=None, dimension=1,
use_prefixes="YZEPTGMkmμnpfazy", format="%.2f pU",
logarithmic=False, log_base=10
):
def __init__(self, value, unit=None, base=10, dimension=1, format="%.2f pU", overshoot=1.5, only_prefixes=[]):
"""
value the numerical value of the variable (int or float)
unit the symbol to use, if any (str or None)
dimension the dimensionality of the value (1, 2, 3, ...)
use_prefixes which multiplier prefixes to use
- the default "YZEPTGMkmμnpfazy" omits "c" for centi- and "D" for deca-
format use printf notation for value (e.g. %010.5f), p for prefix and U for unit
logarithmic when True, log_10 prefixes are used
log_base logarithm base value
param value value to display [int, float]
param unit base unit (e.g. "m") [str]
param base see Base [int]
param dimension dimensionality of the value (1, 2, 3, ...) [int]
param format format string [str]
param overshoot how much should a lower prefix persist [float]
param restrict_prefixes list of whitelisted prefixes; uses everything if empty [list of str]
note that deca- is correctly rendered as "da", the "D" is used in use_prefixes only
# Base
With linear units, this is either exactly 2 for Base 2 prefixes (Ki, Mi, Gi, ...)
or exactly 10 for Base 10 ones (K, M, G, ... as well as m, µ, n, ...).
Negative bases force logarithmic behavior with the base equal to the absolute
value, so e.g. base=-10 forces log10 decibels.
# Overshoot behavior
Overshoot adjusts the boundary between prefixes, e.g. with overshoot=1.0
a Unit switches from k to M when the value exceeds 1000. With overshoot=1.5,
this happens at 1500.
"""
assert (base in [2, 10]) or base < 0
self.value = float(value)
self.unit = unit if unit else ""
self.unit = unit or ""
self.base = base
self.dimension = dimension
self.use_prefixes = use_prefixes
self.format = format
self.logarithmic = logarithmic
self.log_base = log_base
self.overshoot = overshoot
if self.logarithmic and (self.value < 0 or self.log_base < 0):
all_prefixes = self._prefixes_base2 if base == 2 else self._prefixes_base10
if only_prefixes:
self.prefixes = [(prefix, exp) for prefix, exp in all_prefixes if prefix in only_prefixes]
else:
self.prefixes = all_prefixes
if self.base < 0 and self.value < 0:
raise ValueError("math domain error (negative values can't be represented in dB)")
def __str__(self):
if self.value == 0.0:
e = 0
else:
e = round(math.log(abs(self.value), 10), 6)
# each branch determines a prefix and an adjusted value
if self.base > 0:
for prefix, exp in self.prefixes:
if exp == 0: # always prefer 1.0
overshoot = 1
else:
overshoot = self.overshoot
if self.logarithmic:
prefix = "dB"
value = math.log(self.value, self.log_base) * 10
else:
for prefix, mul in self._prefixes:
if prefix in self.use_prefixes and mul*self.dimension <= e:
if self.base ** (exp * self.dimension) * overshoot <= self.value:
break
value = self.value / 10**(mul*self.dimension)
value = self.value / self.base ** (exp * self.dimension)
else:
prefix = "dB"
value = math.log(self.value, -self.base) * 10
output = self.format %(value)
output = output.replace("p", prefix if prefix != "D" else "da") # deca- handler
output = output.replace("p", prefix)
output = output.replace("U", self.unit)
return output
@ -283,35 +297,21 @@ def count_leading(text, char):
# --------------------------------------------------
class Output:
class Log:
class tag:
class short:
info = " "
debug = " <c>?<.>"
start = "<W>>>><.>"
exec = start
note = " <C>#<.>"
warn = " <Y>!<.>"
fail = "<R>!!!<.>"
done = " <G>*<.>"
class long:
info = "INFO"
debug = "<c>DEBG<.>"
start = "<W>EXEC<.>"
exec = start
note = "<C>NOTE<.>"
warn = "<Y>WARN<.>"
fail = "<R>FAIL<.>"
epic = "<R>EPIC<.>"
done = "<G>DONE<.>"
ts = tag.short
tl = tag.long
info = "INFO"
debug = "<b>DEBG<.>"
begin = "<W>BEGN<.>"
note = "<C>NOTE<.>"
warn = "<Y>WARN<.>"
fail = "<R>FAIL<.>"
epic = "<R>EPIC<.>"
done = "<G>DONE<.>"
"""
Text UI output renderer.
Prints messages to the stdout with optional eye candy
Prints messages to stderr with optional eye candy
like colors and timestamps.
If the output stream is not a tty, colors are disabled.
@ -319,13 +319,13 @@ class Output:
Formatting
==========
You can use all formatting characters supported by strftime, as well as:
<T> - tag
<@> - tag
<n> - name
<t> - supplied text
<i> - indentation start marker
"""
def __init__(self, name, format="[%Y-%m-%d %H:%M:%S %Z] <@> -- <n>, <i><t>", colors=True, end=".\n", indent=True, stream=sys.stderr):
def __init__(self, name, format="[%Y-%m-%d %H:%M:%S %Z] <@> -- <n>, <i><t>", colors=True, end="\n", indent=True, stream=sys.stderr):
self.name = name
self.format = format
self.end = end
@ -337,12 +337,12 @@ class Output:
def add_sink(self, stream, colors):
self.sinks.append((stream, colors))
def write(self, text, tag=None, format=None, end=None):
def write(self, fmt, *args, tag=None, format=None, end=None):
"""
@while displaying text
"""
tag = tag or self.__class__.tag.long.info
tag = tag or self.__class__.tag.info
format = format or self.format
end = end or self.end
@ -350,6 +350,8 @@ class Output:
output = output.replace("<@>", tag)
output = output.replace("<n>", self.name)
text = fmt % args
if self.indent and "<i>" in format:
indent_width = render(output, colors=False).index("<i>")
text = paragraph(text, indent=indent_width)[indent_width:]
@ -362,16 +364,30 @@ class Output:
sink.write(render(output, colors))
sink.flush()
def flush(self):
"""
Dummy method for compatibility.
"""
pass
def __call__(self, *args, **kwargs):
def info(self, *args, **kwargs):
self.write(*args, **kwargs)
def debug(self, *args, **kwargs):
self.write(*args, tag=self.tag.debug, **kwargs)
def begin(self, *args, **kwargs):
self.write(*args, tag=self.tag.begin, **kwargs)
def note(self, *args, **kwargs):
self.write(*args, tag=self.tag.note, **kwargs)
def warn(self, *args, **kwargs):
self.write(*args, tag=self.tag.warn, **kwargs)
def fail(self, *args, **kwargs):
self.write(*args, tag=self.tag.fail, **kwargs)
def epic(self, *args, **kwargs):
self.write(*args, tag=self.tag.epic, **kwargs)
def done(self, *args, **kwargs):
self.write(*args, tag=self.tag.done, **kwargs)
# --------------------------------------------------
def get_terminal_size():
@ -391,15 +407,14 @@ def get_terminal_size():
# --------------------------------------------------
class _ProgressBarChannel:
print = Output("over.text._ProgressBarChannel")
def __init__(self, unit, top, precision, use_prefixes=True, prefix_base2=False, min_width_raw=0, min_width_rate=0, min_width_time=0, min_width_percent=7):
def __init__(self, unit, top, precision, use_prefixes=True, base=10, only_prefixes=[], min_width_raw=0, min_width_rate=0, min_width_time=0, min_width_percent=7):
self.unit = unit
self.top = top
self.dynamic_top = top is None
self.prefix_base2 = prefix_base2
self.base = base
self.precision = precision
self.use_prefixes = use_prefixes
self.only_prefixes = only_prefixes
self.min_width = {
"raw": min_width_raw,
"rate": min_width_rate,
@ -410,9 +425,6 @@ class _ProgressBarChannel:
self._value = 0
self.set()
if prefix_base2:
self.print("Unit does not yet support base2 prefixes (e.g. Gi, Mi), using decadic (G, M) instead", self.print.tl.warn)
def set(self, value=None):
if value is not None:
self._value = value
@ -423,10 +435,13 @@ class _ProgressBarChannel:
if unit is None:
unit = self.unit
u = Unit(value, unit, format="%.{:d}f pU".format(self.precision))
if not self.use_prefixes or just == "percent":
u._prefixes = (("", 0),) # Unit needs fixin"
u = Unit(
value,
unit,
format="%.{:d}f pU".format(self.precision),
base=self.base,
only_prefixes=[""] if (not self.use_prefixes or just == "percent") else self.only_prefixes
)
s = str(u)
@ -490,20 +505,18 @@ class ProgressBar:
width is the desired size of the progressbar drawing area in characters (columns)
and defaults to terminal width.
space_before_unit enables a space character between a value and its unit.
channels is a dict that for each channel lists its properties:
{
"a": {
"unit": "f",
"prefix_base2": False,
"top": 39600,
"precision": 1
},
"b": {
"unit": "o",
"prefix_base2": True,
"unit": "B",
"base": 2,
"only_prefixes": ["Ki", "Mi"],
"top": measure_file_size_closure("/path/to/file"),
"precision": 0
}
@ -511,7 +524,7 @@ class ProgressBar:
Channel IDs (the "a", "b") are arbitrary lowercase letters ([a-z]).
Properties "unit" and "prefix_base2" are passed to over.core.text.Unit.
Properties "unit", "base", and "only_prefixes" are passed to over.text.Unit.
"top" is the value of the channel that corresponds to 100%. Channels are
allowed to exceed this value. It can either be a number or a callable. If
it"s a callable, it will be called without arguments and shall return a number.
@ -658,28 +671,64 @@ class ProgressBar:
# --------------------------------------------------
if __name__ == "__main__":
o = Output("over.text")
o("Sample info message")
o("Sample debug message", o.tl.debug)
o("Sample action start message", o.tl.start)
o("Sample action success message", o.tl.done)
o("Sample warning message", o.tl.warn)
o("Sample error message", o.tl.fail)
def timestamp_to_hms(t):
"""
Converts a duration in seconds to a [[?h ]?m ]?s representation.
o("Available colors", end=":\n")
>>> timestamp_to_hms(12345)
'3h 25m 45s'
>>> timestamp_to_hms(3600)
'1h 0m 0s'
>>> timestamp_to_hms(1234)
'20m 34s'
>>> timestamp_to_hms(12)
'12s'
"""
hh = t // 3600
mm = (t % 3600) // 60
ss = t % 60
out = []
if hh:
out.append("%dh" %(hh))
if mm or hh:
out.append("%dm" %(mm))
out.append("%ds" %(ss))
return " ".join(out)
# --------------------------------------------------
if __name__ == "__main__":
log = Log("over.text")
log.info("Sample info message")
log.debug("Sample debug message")
log.begin("Sample action start message")
log.done("Sample action success message")
log.warn("Sample warning message")
log.fail("Sample error message")
log.note("Sample note message")
log.write("Available colors", end=":\n")
for abbr, (code, name) in sorted(ansi_colors.items()):
o("%s = <%s>%s<.>" %(abbr, abbr, name))
log.write("%s = <%s>%s<.>", abbr, abbr, name)
o("ProgressBar test")
log.begin("ProgressBar test")
pb = ProgressBar(
"§%a (§ra/§za) [§=a>§ A] §sa [§rb/§zb] (ETA §TA)",
{
"a": {
"unit": "f",
"prefix_base2": False,
"base": 10,
"top": 39600,
"precision": 1,
"min_width_raw": 10,
@ -688,8 +737,8 @@ if __name__ == "__main__":
"min_width_time": 0
},
"b": {
"unit": "o",
"prefix_base2": True,
"unit": "B",
"base": 2,
"top": 3200,
"precision": 0,
"min_width_raw": 10,
@ -707,4 +756,4 @@ if __name__ == "__main__":
pb.render()
time.sleep(0.25)
pb.end(not True)
pb.end(True)

View file

@ -20,13 +20,15 @@ class ndict:
{"alpha": 1, "beta": 42}
"""
__methods__ = ["values", "items"]
__methods__ = ["values", "items", "keys"]
def __init__(self, *args, **kwargs):
object.__setattr__(self, "d", OrderedDict(*args, **kwargs))
def __repr__(self):
return "|" + repr(self.d)
atoms = ["(%s, %s)" %(repr(k), repr(v)) for k, v in self.items()]
return "ndict([" + ", ".join(atoms) + "])"
def __iter__(self):
return self.d.__iter__()

View file

@ -1,8 +1,8 @@
#! /usr/bin/env python3
# encoding: utf-8
major = 2 # VERSION_MAJOR_IDENTIFIER
major = 3 # VERSION_MAJOR_IDENTIFIER
minor = 0 # VERSION_MINOR_IDENTIFIER
# VERSION_LAST_MM 2.0
# VERSION_LAST_MM 3.0
patch = 0 # VERSION_PATCH_IDENTIFIER
str = "2.0.0" # VERSION_STRING_IDENTIFIER
str = "3.0.0" # VERSION_STRING_IDENTIFIER