def __strict_strtobool(value):
from distutils.util import strtobool
if isinstance(value, bool):
return value
try:
lower_text = value.lower()
except AttributeError:
raise ValueError("invalid value '{}'".format(str(value)))
binary_value = strtobool(lower_text)
if lower_text not in ["true", "false"]:
raise ValueError("invalid value '{}'".format(str(value)))
return bool(binary_value)
python类strtobool()的实例源码
def boolify(value):
""" Function that will translate common strings into bool values
True -> "True", "t", "yes", "y", "on", "1"
False -> any other string
Objects other than string will be transformed
using built-in bool() function.
"""
if isinstance(value, basestring):
try:
return bool(strtobool(value))
except ValueError:
return False
else:
return bool(value)
def boolify(value):
""" Function that will translate common strings into bool values
True -> "True", "t", "yes", "y", "on", "1"
False -> any other string
Objects other than string will be transformed
using built-in bool() function.
"""
if isinstance(value, basestring):
try:
return bool(strtobool(value))
except ValueError:
return False
else:
return bool(value)
def logging_levels():
"""
Context manager to conditionally set logging levels.
Supports setting per-request debug logging using the `X-Request-Debug` header.
"""
enabled = strtobool(request.headers.get("x-request-debug", "false"))
level = None
try:
if enabled:
level = getLogger().getEffectiveLevel()
getLogger().setLevel(DEBUG)
yield
finally:
if enabled:
getLogger().setLevel(level)
def config(self, config):
"""VES option configuration"""
for key, value in config.items('config'):
if key in self._app_config:
try:
if type(self._app_config[key]) == int:
value = int(value)
elif type(self._app_config[key]) == float:
value = float(value)
elif type(self._app_config[key]) == bool:
value = bool(strtobool(value))
if isinstance(value, type(self._app_config[key])):
self._app_config[key] = value
else:
logging.error("Type mismatch with %s" % key)
sys.exit()
except ValueError:
logging.error("Incorrect value type for %s" % key)
sys.exit()
else:
logging.error("Incorrect key configuration %s" % key)
sys.exit()
def __init__(self, obj, config=None):
_config = {FILEONLY: False}
if config is not None:
_configlist = [c.strip().lower() for c in config.split('=')]
if len(_configlist) == 2 and _configlist[0] == FILEONLY:
_config[FILEONLY] = strtobool(_configlist[1])
else:
raise ValueError('Artefact method plugin (used in user defined class %s) accepts a single argument of either '
'"fileonly = True" or "fileonly = False". By default fileonly mode is False '
'and does not need to be specified' % striptoclassname(type(obj)))
super(Artefacts, self).__init__(obj, config=_config)
self.computable = True
def main():
"""JVC Projector tools main menu"""
while True:
try:
Menu()
break
except Exception as err:
if isinstance(err, ExceptionInThread):
err, exc = err.args
else:
exc = sys.exc_info()
print(err)
try:
if strtobool(input('error occured print stack trace? ')):
traceback.print_exception(*exc)
except:
pass
try:
if not strtobool(input('restart? ')):
break
except:
break
def humanbool(name, value):
"""
Determine human boolean value
:Parameters:
`name` : ``str``
The config key (used for error message)
`value` : ``str``
The config value
:Return: The boolean value
:Rtype: ``bool``
:Exceptions:
- `ValueError` : The value could not be recognized
"""
try:
return _util.strtobool(str(value).strip().lower() or 'no')
except ValueError:
raise ValueError("Unrecognized config value: %s = %s" % (name, value))
def humanbool(name, value):
"""
Determine human boolean value
:Parameters:
`name` : ``str``
The config key (used for error message)
`value` : ``str``
The config value
:Return: The boolean value
:Rtype: ``bool``
:Exceptions:
- `ValueError` : The value could not be recognized
"""
try:
return _util.strtobool(str(value).strip().lower() or 'no')
except ValueError:
raise ValueError("Unrecognized config value: %s = %s" % (name, value))
def humanbool(name, value):
"""
Determine human boolean value
:Parameters:
`name` : ``str``
The config key (used for error message)
`value` : ``str``
The config value
:Return: The boolean value
:Rtype: ``bool``
:Exceptions:
- `ValueError` : The value could not be recognized
"""
try:
return _util.strtobool(str(value).strip().lower() or 'no')
except ValueError:
raise ValueError("Unrecognized config value: %s = %s" % (name, value))
def humanbool(name, value):
"""
Determine human boolean value
:Parameters:
`name` : ``str``
The config key (used for error message)
`value` : ``str``
The config value
:Return: The boolean value
:Rtype: ``bool``
:Exceptions:
- `ValueError` : The value could not be recognized
"""
try:
return _util.strtobool(str(value).strip().lower() or 'no')
except ValueError:
raise ValueError("Unrecognized config value: %s = %s" % (name, value))
def _str_to_bool(self, val):
try:
return bool(strtobool(val))
except ValueError:
return None
def update_defaults(self, defaults):
"""Updates the given defaults with values from the config files and
the environ. Does a little special handling for certain types of
options (lists)."""
# Then go and look for the other sources of configuration:
config = {}
# 1. config files
for section in ('global', self.name):
config.update(self.normalize_keys(self.get_config_section(section)))
# 2. environmental variables
config.update(self.normalize_keys(self.get_environ_vars()))
# Then set the options with those values
for key, val in config.items():
option = self.get_option(key)
if option is not None:
# ignore empty values
if not val:
continue
if option.action in ('store_true', 'store_false', 'count'):
val = strtobool(val)
if option.action == 'append':
val = val.split()
val = [self.check_default(option, key, v) for v in val]
else:
val = self.check_default(option, key, val)
defaults[option.dest] = val
return defaults
def boolify(value):
""" Function that will translate common strings into bool values
True -> "True", "t", "yes", "y", "on", "1"
False -> any other string
Objects other than string will be transformed using built-in bool() function.
"""
if isinstance(value, basestring):
try:
return bool(strtobool(value))
except ValueError:
return False
else:
return bool(value)
def update_defaults(self, defaults):
"""Updates the given defaults with values from the config files and
the environ. Does a little special handling for certain types of
options (lists)."""
# Then go and look for the other sources of configuration:
config = {}
# 1. config files
for section in ('global', self.name):
config.update(self.normalize_keys(self.get_config_section(section)))
# 2. environmental variables
config.update(self.normalize_keys(self.get_environ_vars()))
# Then set the options with those values
for key, val in config.items():
option = self.get_option(key)
if option is not None:
# ignore empty values
if not val:
continue
if option.action in ('store_true', 'store_false', 'count'):
val = strtobool(val)
if option.action == 'append':
val = val.split()
val = [self.check_default(option, key, v) for v in val]
else:
val = self.check_default(option, key, val)
defaults[option.dest] = val
return defaults
def boolify(value):
""" Function that will translate common strings into bool values
True -> "True", "t", "yes", "y", "on", "1"
False -> any other string
Objects other than string will be transformed using built-in bool() function.
"""
if isinstance(value, basestring):
try:
return bool(strtobool(value))
except ValueError:
return False
else:
return bool(value)
def should_skip_logging(func):
"""
Should we skip logging for this handler?
"""
disabled = strtobool(request.headers.get("x-request-nolog", "false"))
return disabled or getattr(func, SKIP_LOGGING, False)
def configure_audit_decorator(graph):
"""
Configure the audit decorator.
Example Usage:
@graph.audit
def login(username, password):
...
"""
include_request_body = int(graph.config.audit.include_request_body)
include_response_body = int(graph.config.audit.include_response_body)
include_path = strtobool(graph.config.audit.include_path)
include_query_string = strtobool(graph.config.audit.include_query_string)
def _audit(func):
@wraps(func)
def wrapper(*args, **kwargs):
options = AuditOptions(
include_request_body=include_request_body,
include_response_body=include_response_body,
include_path=include_path,
include_query_string=include_query_string,
)
return _audit_request(options, func, graph.request_context, *args, **kwargs)
return wrapper
return _audit
def to_bool(env, default='false'):
"""
Convert a string to a bool.
"""
return bool(util.strtobool(os.getenv(env, default)))
def __init__(self, os_auth_args):
"""Initialization method for class.
:param os_auth_args: dict containing auth creds.
:type os_auth_args: dict
"""
self.os_auth_args = os_auth_args
insecure = bool(strtobool(self.os_auth_args.get('insecure', 'False')))
self.verify = insecure is False
def str2bool(s):
if is_string(s):
return strtobool(s)
return bool(s)
def __init__(self, file):
if python_3():
self.__settings = configparser.ConfigParser()
else:
self.__settings = ConfigParser.SafeConfigParser()
try:
with open(file) as f:
self.__settings.readfp(f)
except IOError as e:
raise ConfigfileException(e)
# Main
self.data['main'] = {'customMethods': self.__settings.get('Main', 'Custom_Methods'),
'log_level': self.__settings.get('Main', 'Log_Level'),
'daemon': strtobool(self.__settings.get('Main', 'Daemon')),
'update_rate': int(self.__settings.get('Main', 'Config_Updaterate_in_Minutes')) * 60,
'log_performance': strtobool(self.__settings.get('Main', 'Log_Performance'))}
# Livestatus
livestatus_split = self.__settings.get('Livestatus', 'Address').split(":")
self.data['livestatus'] = {'protocol': livestatus_split[0], 'address': livestatus_split[1]}
if len(livestatus_split) == 3:
self.data['livestatus']['port'] = int(livestatus_split[2])
# Histou
# self.data['histou'] = {'prot': "http", 'address': self.__settings.get('Histou', 'Address')}
histou_split = self.__settings.get('Histou', 'Address').split(":", 1)
self.data['histou'] = {'prot': histou_split[0], 'address': histou_split[1]}
self.data['histou']['user'] = self.__settings.get('Histou', 'User')
self.data['histou']['password'] = self.__settings.get('Histou', 'Password')
# Influxdb
self.data['influxdb'] = {'read': {'address': self.__settings.get('InfluxDB', 'Address_Read'),
'db': self.__settings.get('InfluxDB', 'DB_Read'),
'args': self.__settings.get('InfluxDB', 'DB_Read_Args')},
'write': {'address': self.__settings.get('InfluxDB', 'Address_Write'),
'db_forecast': self.__settings.get('InfluxDB', 'DB_Write_Forecast'),
'db_anomaly': self.__settings.get('InfluxDB', 'DB_Write_Anomaly'),
'args': self.__settings.get('InfluxDB', 'DB_Write_Args')}}
def __init__(self):
_scheme = runtimenamespace().get('outputscheme', None)
if _scheme is None:
raise ValueError('Output scheme is not defined')
_outputconfig = partialnamespace('output').get(_scheme, None)
if _outputconfig is None:
raise ValueError('Output scheme configuration was not retrieved')
_autolink = _outputconfig.get('autolink', None)
if _autolink is not None:
self.autolink = strtobool(_autolink)
else:
self.autolink = True
_linkattribute = _outputconfig.get('link', None)
if _linkattribute is None and not self.autolink:
raise ValueError('a "link:" value must be set in [output.%s] '
'in gtool.cfg when using directedgraph output '
'and autolink is disabled' % _scheme)
self.linkattribute = _linkattribute
_linkproperties = _outputconfig.get('linkprops', None)
if _linkproperties is not None:
self.linkproperties = [l.strip() for l in _linkproperties.split(',')]
else:
self.linkproperties = []
super(Directedgraph, self).__init__()
def __validate__(self, valuedict):
_public = valuedict.get('public', None)
if _public is None:
public = 0 # False
else:
public = strtobool('%s' % _public)
warnings.warn('public check does not currently work due a problem in the underlying validators library')
try:
url(self.__value__, public=public) # TODO public / private test doesn't work
except ValidationFailure:
raise ValueError('Was expecting a valid %s but got %s' % ('public URL' if public else 'URL', len(self.__value__)))
return True
def __validate__(self, valuedict):
_public = valuedict.get('public', None)
if _public is None:
public = 0 # False
else:
public = strtobool('%s' % _public)
warnings.warn('public check does not currently work due a problem in the underlying validators library')
try:
url(self.__value__, public=public) # TODO public / private test doesn't work
except ValidationFailure:
raise ValueError('Was expecting a valid %s but got %s' % ('public URL' if public else 'URL', len(self.__value__)))
return True
def start_backup(self):
'''
Read settings and starts backup
'''
for section in self._config.sections():
# Run a backup for each section
print "Starting backup for pool {}".format(section)
images = self.getsetting(section, 'images').split(',')
backup_dest = self.getsetting(section, 'destination directory')
conf_file = self.getsetting(section, 'ceph config')
check_mode = bool(strtobool(self.getsetting(section, 'check mode')))
compress_mode = bool(strtobool(self.getsetting(section, 'compress')))
window_size = int(self.getsetting(section, 'window size'))
window_unit = self.getsetting(section, 'window unit')
backup_mode = self.getsetting(section, 'backup mode')
cb = CephFullBackup(section, images, backup_dest, conf_file, check_mode, compress_mode, window_size, window_unit)
if backup_mode == 'full':
print "Full ceph backup"
cb.print_overview()
cb.full_backup()
elif backup_mode == 'incremental':
print "Incremental ceph backup"
cb.print_overview()
cb.incremental_backup()
else:
raise Exception("Unknown backup mode: {}".format(backup_mode))
def update_defaults(self, defaults):
"""Updates the given defaults with values from the config files and
the environ. Does a little special handling for certain types of
options (lists)."""
# Then go and look for the other sources of configuration:
config = {}
# 1. config files
for section in ('global', self.name):
config.update(self.normalize_keys(self.get_config_section(section)))
# 2. environmental variables
config.update(self.normalize_keys(self.get_environ_vars()))
# Then set the options with those values
for key, val in config.items():
option = self.get_option(key)
if option is not None:
# ignore empty values
if not val:
continue
if option.action in ('store_true', 'store_false', 'count'):
val = strtobool(val)
if option.action == 'append':
val = val.split()
val = [self.check_default(option, key, v) for v in val]
else:
val = self.check_default(option, key, val)
defaults[option.dest] = val
return defaults
def headless_mode():
return not strtobool(os.environ.get('HEADED', 'False'))
def cast_string_to_type(
value: str,
cast: [str, bool, int]
) -> [str, bool, int]:
if cast is str:
return value
elif cast is bool:
return strtobool(value)
elif cast is int:
return int(value)
else:
raise KeyError('Invalid `cast` param')
def __init__(self, feat_stride='16', scales='(8, 16, 32)', ratios='(0.5, 1, 2)', output_score='False',
rpn_pre_nms_top_n='6000', rpn_post_nms_top_n='300', threshold='0.3', rpn_min_size='16'):
super(ProposalProp, self).__init__(need_top_grad=False)
self._feat_stride = int(feat_stride)
self._scales = scales
self._ratios = ratios
self._output_score = strtobool(output_score)
self._rpn_pre_nms_top_n = int(rpn_pre_nms_top_n)
self._rpn_post_nms_top_n = int(rpn_post_nms_top_n)
self._threshold = float(threshold)
self._rpn_min_size = int(rpn_min_size)
