From 504f0155ad6f91536085e27f63f4d178af9591ab Mon Sep 17 00:00:00 2001 From: jesko Date: Fri, 11 Mar 2022 20:13:04 +0100 Subject: [PATCH] puts evar-config into module, adds band-aid bypass --- refinery/explore.py | 3 +- refinery/lib/environment.py | 152 ++++++++++++++++++++++++++++++++++++ refinery/lib/powershell.py | 25 +++--- refinery/lib/tools.py | 8 +- refinery/units/__init__.py | 70 ++--------------- run-tests.py | 4 +- 6 files changed, 181 insertions(+), 81 deletions(-) create mode 100644 refinery/lib/environment.py diff --git a/refinery/explore.py b/refinery/explore.py index 4ae758f1a7..74f4141c97 100644 --- a/refinery/explore.py +++ b/refinery/explore.py @@ -39,8 +39,9 @@ def get_help_string(unit, brief=False, width=None): else: from io import StringIO from os import environ + from refinery.lib.environment import environment try: - environ['REFINERY_TERMSIZE'] = str(width) + environ[environment.term_size.key] = str(width) argp = unit.argparser() except ArgparseError as fail: argp = fail.parser diff --git a/refinery/lib/environment.py b/refinery/lib/environment.py new file mode 100644 index 0000000000..ef92ed0482 --- /dev/null +++ b/refinery/lib/environment.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +A common interface to all binary refinery configuration settings available via environment +variables. This module is also host to the logging configuration. +""" +from __future__ import annotations + +import os +import logging + +from enum import IntEnum +from typing import Any, Optional + + +class LogLevel(IntEnum): + """ + An enumeration representing the current log level: + """ + DETACHED = logging.CRITICAL + 100 + """ + This unit is not attached to a terminal but has been instantiated in + code. This means that the only way to communicate problems is to throw + an exception. + """ + NONE = logging.CRITICAL + 50 + + @classmethod + def FromVerbosity(cls, verbosity: int): + if verbosity < 0: + return cls.DETACHED + return { + 0: cls.WARNING, + 1: cls.INFO, + 2: cls.DEBUG + }.get(verbosity, cls.DEBUG) + + NOTSET = logging.NOTSET # noqa + CRITICAL = logging.CRITICAL # noqa + FATAL = logging.FATAL # noqa + ERROR = logging.ERROR # noqa + WARNING = logging.WARNING # noqa + WARN = logging.WARN # noqa + INFO = logging.INFO # noqa + DEBUG = logging.DEBUG # noqa + + @property + def verbosity(self) -> int: + if self.value >= LogLevel.DETACHED: + return -1 + if self.value >= LogLevel.WARNING: + return +0 + if self.value >= LogLevel.INFO: + return +1 + if self.value >= LogLevel.DEBUG: + return +2 + else: + return -1 + + +logging.addLevelName(logging.CRITICAL, 'failure') # noqa +logging.addLevelName(logging.ERROR, 'failure') # noqa +logging.addLevelName(logging.WARNING, 'warning') # noqa +logging.addLevelName(logging.INFO, 'comment') # noqa +logging.addLevelName(logging.DEBUG, 'verbose') # noqa + + +def logger(name: str) -> logging.Logger: + """ + Obtain a logger which is configured with the default refinery format. + """ + logger = logging.getLogger(name) + if not logger.hasHandlers(): + stream = logging.StreamHandler() + stream.setFormatter(logging.Formatter( + '({asctime}) {levelname} in {name}: {message}', + style='{', + datefmt='%H:%M:%S' + )) + logger.addHandler(stream) + logger.propagate = False + return logger + + +class EnvironmentVariableSetting: + key: str + val: Any + + def __init__(self, name: str): + self.key = F'REFINERY_{name}' + self.val = self.read() + + @property + def value(self): + return self.val + + def read(self): + return None + + +class EVBool(EnvironmentVariableSetting): + val: bool + + def read(self): + value = os.environ.get(self.key, None) + if value is None: + return False + else: + value = value.lower().strip() + if not value: + return False + if value.isdigit(): + return bool(int(value)) + return value not in {'no', 'off', 'false'} + + +class EVInt(EnvironmentVariableSetting): + val: int + + def read(self) -> int: + try: + return int(os.environ[self.key], 0) + except (KeyError, ValueError): + return 0 + + +class EVLog(EnvironmentVariableSetting): + val: Optional[LogLevel] + + def read(self): + try: + loglevel = os.environ[self.key] + except KeyError: + return None + if loglevel.isdigit(): + return LogLevel.FromVerbosity(int(loglevel)) + try: + loglevel = LogLevel[loglevel] + except KeyError: + levels = ', '.join(ll.name for ll in LogLevel) + logger(__name__).warning( + F'ignoring unknown verbosity "{loglevel!r}"; pick from: {levels}') + return None + else: + return loglevel + + +class environment: + verbosity = EVLog('VERBOSITY') + term_size = EVInt('TERM_SIZE') + silence_ps1_warning = EVBool('SILENCE_PS1_WARNING') + disable_ps1_bandaid = EVBool('DISABLE_PS1_BANDAID') diff --git a/refinery/lib/powershell.py b/refinery/lib/powershell.py index 3f63651832..578b4f3e6a 100644 --- a/refinery/lib/powershell.py +++ b/refinery/lib/powershell.py @@ -8,6 +8,8 @@ import ctypes import os +from refinery.lib.environment import environment + _PS1_MAGIC = B'[BRPS1]:' @@ -112,19 +114,16 @@ def write(self, data): if not self._header_written: self.stream.write(_PS1_MAGIC) self._header_written = True - if not Ps1Wrapper.WRAPPED: - EV = 'REFINERY_SUPPRESS_PS1_WARNING' - ev = os.environ.get(EV, '0') - ev = int(ev) if ev.isdigit() else bool(ev) - if not ev: - import logging - logging.getLogger('root').critical( - U'WARNING: PowerShell has no support for binary pipelines or streaming. Binary Refinery ' - U'uses an unreliable and slow workaround: It is strongly recommended to use the command ' - U'processor instead. Proceed at your own peril!\n' - U'- To get more information: https://github.com/binref/refinery/issues/5\n' - F'- To disable this warning: $env:{EV}=1' - ) + if not Ps1Wrapper.WRAPPED and not environment.silence_ps1_warning.value: + import logging + logging.getLogger('root').critical( + U'WARNING: PowerShell has no support for binary pipelines or streaming. Binary Refinery ' + U'uses an unreliable and slow workaround: It is strongly recommended to use the command ' + U'processor instead. Proceed at your own peril!\n' + F'- To silence this warning: $env:{environment.silence_ps1_warning.key}=1\n' + F'- To disable the band-aid: $env:{environment.disable_ps1_bandaid.key}=1\n' + U'- To get more information: https://github.com/binref/refinery/issues/5' + ) view = memoryview(data) size = 1 << 15 for k in range(0, len(view), size): diff --git a/refinery/lib/tools.py b/refinery/lib/tools.py index 1c18b5e2d6..978df9cabe 100644 --- a/refinery/lib/tools.py +++ b/refinery/lib/tools.py @@ -54,10 +54,10 @@ def get_terminal_size(default=0): of the terminal cannot be determined of if the width is less than 8 characters, the function returns zero. """ - try: - return int(os.environ['REFINERY_TERMSIZE']) - except (KeyError, ValueError): - pass + from refinery.lib.environment import environment + ev_terminal_size = environment.term_size.value + if ev_terminal_size > 0: + return ev_terminal_size width = default for stream in (sys.stderr, sys.stdout): if stream.isatty(): diff --git a/refinery/units/__init__.py b/refinery/units/__init__.py index 750b49835f..24b005b86d 100644 --- a/refinery/units/__init__.py +++ b/refinery/units/__init__.py @@ -161,7 +161,7 @@ def reverse(self, data): import sys from abc import ABCMeta -from enum import IntEnum, Enum +from enum import Enum from functools import wraps from collections import OrderedDict @@ -198,6 +198,7 @@ def reverse(self, data): from refinery.lib.tools import documentation, isstream, lookahead, autoinvoke, one, skipfirst, isbuffer from refinery.lib.frame import Framed, Chunk from refinery.lib.structures import MemoryFile +from refinery.lib.environment import LogLevel, environment class RefineryPartialResult(ValueError): @@ -897,51 +898,6 @@ def logger(cls) -> logging.Logger: return logger -class LogLevel(IntEnum): - """ - An enumeration representing the current log level: - """ - DETACHED = logging.CRITICAL + 100 - """ - This unit is not attached to a terminal but has been instantiated in - code. This means that the only way to communicate problems is to throw - an exception. - """ - NONE = logging.CRITICAL + 50 - - @classmethod - def FromVerbosity(cls, verbosity: int): - if verbosity < 0: - return cls.DETACHED - return { - 0: cls.WARNING, - 1: cls.INFO, - 2: cls.DEBUG - }.get(verbosity, cls.DEBUG) - - NOTSET = logging.NOTSET # noqa - CRITICAL = logging.CRITICAL # noqa - FATAL = logging.FATAL # noqa - ERROR = logging.ERROR # noqa - WARNING = logging.WARNING # noqa - WARN = logging.WARN # noqa - INFO = logging.INFO # noqa - DEBUG = logging.DEBUG # noqa - - @property - def verbosity(self) -> int: - if self.value >= LogLevel.DETACHED: - return -1 - if self.value >= LogLevel.WARNING: - return +0 - if self.value >= LogLevel.INFO: - return +1 - if self.value >= LogLevel.DEBUG: - return +2 - else: - return -1 - - class DelayedArgumentProxy: """ This class implements a proxy for the `args` member variable of `refinery.units.Unit`. @@ -1746,8 +1702,11 @@ def run(cls: Union[Type[Unit], Executable], argv=None, stream=None) -> None: this method will be executed when a class inheriting from `refinery.units.Unit` is defined in the current `__main__` module. """ - from refinery.lib import powershell - ps1 = powershell.bandaid(cls.codec) + if not environment.disable_ps1_bandaid.value: + from refinery.lib import powershell + ps1 = powershell.bandaid(cls.codec) + else: + ps1 = None argv = argv if argv is not None else sys.argv[1:] start_clock = None @@ -1778,20 +1737,7 @@ def run(cls: Union[Type[Unit], Executable], argv=None, stream=None) -> None: if ps1: unit.log_debug(F'applying PowerShell band-aid for: {unit.name}') - try: - loglevel = os.environ['REFINERY_VERBOSITY'] - except KeyError: - loglevel = None - else: - if loglevel.isdigit(): - loglevel = LogLevel.FromVerbosity(int(loglevel)) - else: - try: - loglevel = LogLevel[loglevel] - except KeyError: - levels = ', '.join(ll.name for ll in LogLevel) - unit.log_warn(F'unknown verbosity {loglevel!r}, pick from {levels}') - loglevel = None + loglevel = environment.verbosity.value if loglevel: unit.log_level = loglevel diff --git a/run-tests.py b/run-tests.py index 4e56a892e6..6b14c77979 100755 --- a/run-tests.py +++ b/run-tests.py @@ -9,6 +9,8 @@ import sys from inspect import stack +from refinery.lib.environment import environment, LogLevel + here = os.path.dirname(os.path.abspath(stack()[0][1])) argp = argparse.ArgumentParser() @@ -17,7 +19,7 @@ args = argp.parse_args() os.chdir('test') -os.environ['REFINERY_VERBOSITY'] = 'DETACHED' +os.environ[environment.verbosity.key] = LogLevel.DETACHED.name suite = unittest.TestLoader().discover('test', F'test_*{args.pattern}*') tests = unittest.TextTestRunner(verbosity=2)