Skip to content

Commit

Permalink
puts evar-config into module, adds band-aid bypass
Browse files Browse the repository at this point in the history
  • Loading branch information
huettenhain committed Mar 11, 2022
1 parent 5245e3b commit 504f015
Show file tree
Hide file tree
Showing 6 changed files with 181 additions and 81 deletions.
3 changes: 2 additions & 1 deletion refinery/explore.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
152 changes: 152 additions & 0 deletions refinery/lib/environment.py
Original file line number Diff line number Diff line change
@@ -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')
25 changes: 12 additions & 13 deletions refinery/lib/powershell.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@
import ctypes
import os

from refinery.lib.environment import environment

_PS1_MAGIC = B'[BRPS1]:'


Expand Down Expand Up @@ -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):
Expand Down
8 changes: 4 additions & 4 deletions refinery/lib/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down
70 changes: 8 additions & 62 deletions refinery/units/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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`.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down
4 changes: 3 additions & 1 deletion run-tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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)
Expand Down

0 comments on commit 504f015

Please sign in to comment.