#
# Command-line interface script to feature flags
#
import argparse
import os
import re
import sys
import pymmlibs
pymmlibs.mmerr_set_mmlibs()
from schrodinger.utils import fileutils # isort:skip # noqa:E402
from schrodinger.utils import log # isort:skip # noqa:E402
from schrodinger.utils import mmutil # isort:skip # noqa:E402
from schrodinger.utils import subprocess # isort:skip # noqa:E402
from schrodinger.utils.featureflags.write import write_features_json, read_json_data # isort:skip # noqa:E402
logger = log.get_output_logger('feature_flags')
CONFLICTING_ENV_VAR = "SCHRODINGER_PERL_FEATURE_FLAGS_ENABLED"
SCHRODINGER_FEATURE_FLAGS = "SCHRODINGER_FEATURE_FLAGS"
GUI_OPTS = {'gui'}
SET_OPTS = {'enable', 'disable'}
LIST_OPTS = {'list', 'features'}
CMDLINE_OPTS = SET_OPTS.union(LIST_OPTS)
[docs]def get_features():
"""
Get the feature flags from site and user state file based
on the existence.
:rtype: dict
:return: Dictionary of feature name and its corresponding json item.
"""
feature_flags = {}
site_feature_file = mmutil.get_feature_flags_site_state_file()
if site_feature_file:
feature_flags = read_json_data(site_feature_file)
site_features = set(list(feature_flags))
else:
site_features = set()
user_feature_flags = read_json_data(
mmutil.get_feature_flags_user_state_file())
user_features = set(list(user_feature_flags))
site_features = site_features.difference(user_features)
for k in user_feature_flags:
if k not in feature_flags:
feature_flags[k] = user_feature_flags[k]
return site_features, user_features, feature_flags
[docs]def set_features(features, enable):
"""
Enable or Disable the given features.
This also report the given features which are
not present in the default state file.
:type features: list or set
:param features: List of features to enable or disable in the user state
file.
:returntype: tuple(int, features)
:return: (number of features changed, list of unknown features)
:raise ValueError: if json document cannot be parsed.
"""
featureflags = read_json_data(mmutil.get_feature_flags_user_state_file())
not_available_features = []
count = 0
for feature in features:
default_state = mmutil.feature_flag_default_state_s(feature)
user_state = mmutil.feature_flag_user_state_s(feature)
if default_state != mmutil.FEATURE_NOT_PRESENT:
if feature in featureflags: # Update existing feature
feature_item = featureflags[feature]
# Same as current state already in user file
if feature_item["Enabled"] == enable:
continue
feature_item["Enabled"] = enable
# Same as the state that is in site config file, don't set in user file
elif user_state == enable:
continue
# Same as the state that in default file, don't set in user file
elif default_state == enable:
continue
else: # Add new entry
featureflags[feature] = {"Feature": feature, "Enabled": enable}
count = count + 1
else:
not_available_features.append(feature)
if count:
feature_flags_format = list(featureflags.values())
user_state_file = mmutil.get_feature_flags_user_state_file()
write_features_json(feature_flags_format, user_state_file)
return (count, not_available_features)
[docs]def get_state_string(state):
"""
Convert the feature state from integer to a string.
:type state: int
:param state: Feature state in integer format.
:returntype: string
:return: 0 => Disabled, 1 => Enabled, -1 => None
"""
if state == mmutil.FEATURE_DISABLED:
state_str = "Disabled"
elif state == mmutil.FEATURE_ENABLED:
state_str = "Enabled"
elif state == mmutil.FEATURE_NOT_PRESENT:
state_str = "None"
else:
state_str = "unknown"
return state_str
[docs]def get_env_var_warning():
"""
:return: If there are conflicting env var settings, return a string with
the description. Otherwise, return None.
:rtype str or None:
"""
if CONFLICTING_ENV_VAR not in os.environ:
return None
env_value = os.environ.get(CONFLICTING_ENV_VAR)
warning = ('WARNING: You have {env_var} set to "{env_value}" in your '
'environment. This overrides your user settings and should be '
'unset.'.format(
env_var=CONFLICTING_ENV_VAR, env_value=env_value))
perl_features = set(env_value.split())
nondefault_jobcontrol_features = {
i
for i in mmutil.feature_flags_get_nondefault_map()
} & set(mmutil.JOBCONTROL_FEATURE_FLAGS)
# Warn about features in the env var that are nondefault jobcontrol if
# they are disabled.
for feature in perl_features & nondefault_jobcontrol_features:
if not mmutil.feature_flag_is_enabled_s(feature):
return warning
# Warn about features not in the env var that are nondefault jobcontrol
# if they are enabled.
for feature in nondefault_jobcontrol_features:
if mmutil.feature_flag_is_enabled_s(feature):
if feature not in perl_features:
return warning
return None
[docs]def env_var_feature_flag_is_set():
"""
Return boolean whether SCHRODINGER_FEATURE_FLAGS is set or not
"""
return SCHRODINGER_FEATURE_FLAGS in os.environ
[docs]def list_features(search="*"):
"""
Show all available features from site state file or user state file
whichever is accessible when search string is not specified. If search
string is given, feature matching the search string will be shown.
Otherwise feature whose description match the search string is shown only
when state file is accessible.
The function will list the output in following format::
<feature_name> : <Enabled|Disabled>
<Description if available>
<User State> : <Enabled|Disabled|None>
<Default State> : <Enabled|Disabled>
:type search_string: str
:param search_string: Optional search string.
"""
unknown = []
site_features, user_features, featureflags = get_features()
if search in featureflags:
featureflags = {search: featureflags[search]}
elif search != '*':
_featureflags = {}
for (feature, item) in featureflags.items():
if "Description" in item:
description = item["Description"]
if re.search(search, description, re.IGNORECASE) is None:
continue
_featureflags[feature] = item
featureflags = _featureflags
if not _featureflags:
all_state = mmutil.feature_flag_default_state_s(search)
if all_state != mmutil.FEATURE_NOT_PRESENT:
featureflags = {search: ""}
else:
featureflags = {}
unknown.append(search)
for (feature, item) in featureflags.items():
all_state = mmutil.feature_flag_is_enabled_s(feature)
default_state = mmutil.feature_flag_default_state_s(feature)
user_state = mmutil.feature_flag_user_state_s(feature)
source_value = "Built-in default"
if feature in site_features:
source_value = mmutil.get_feature_flags_site_state_file()
if env_var_feature_flag_is_set():
if os.environ[SCHRODINGER_FEATURE_FLAGS] == "0":
source_value = "Built-in default"
if feature in os.environ[SCHRODINGER_FEATURE_FLAGS]:
source_value = "Environment variable SCHRODINGER_FEATURE_FLAGS"
elif feature in user_features:
source_value = mmutil.get_feature_flags_user_state_file()
if all_state != mmutil.FEATURE_NOT_PRESENT:
logger.info("{} : {}".format(feature, get_state_string(all_state)))
if item and "Description" in item:
logger.info(" " + item["Description"])
logger.info(" User State : %s" % get_state_string(user_state))
logger.info(
" Default State : %s" % get_state_string(default_state))
logger.info(" Source : %s" % source_value)
logger.info('')
else:
unknown.append(feature)
if unknown:
logger.warning("WARNING: No such feature in default state file - "
"%s" % ' '.join(unknown))
if not featureflags and search == '*':
logger.info("No feature flags are set.")
[docs]def get_user_features():
"""
Return the changes in the user feature flags with respect to the site
feature flags, in the format::
+FLAG1 +FLAG2 -FLAG3 ...
"""
modified_features = []
for feature in mmutil.feature_flags_get_nondefault_map():
enabled = mmutil.feature_flag_is_enabled_s(feature)
if enabled:
modified_features.append("+" + feature)
else:
modified_features.append("-" + feature)
return " ".join(modified_features)
[docs]def parse_args(argv=None):
"""
Setup code for argument parsing
:type argv: list
:params argv: Arguments to parse.
"""
usage = '''
feature_flags.py [-h] [-e <feature> [<feature> ...] -d <feature>
[<feature> ...]] | [-l [<feature>]] | [-g]
'''
parser = argparse.ArgumentParser(
description="Command-line interface to feature flags",
usage=usage,
epilog="If no feature is provided, the program will list the "
"state of all features from user state file and will "
"indicate the state explicity set by the user "
"that are equal to the default value.")
parser.add_argument(
'-e',
'--enable',
metavar='<feature>',
dest='enable',
nargs='+',
help="Enable the given feature(s) in the user state file.")
parser.add_argument(
'-d',
'--disable',
metavar='<feature>',
dest='disable',
nargs='+',
help="Disable the given feature(s) in the user state file.")
group_ex = parser.add_mutually_exclusive_group()
group_ex.add_argument(
'-l',
'--list',
metavar='<feature>',
nargs='?',
const='*',
help="When no search string is specified, show all available "
"feature flags with descriptions from the default "
"feature file or user state file whichever is accessible. "
"If search string is provided, we list the details of the "
"feature if it is a feature name. Otherwise feature flags "
"and their descriptions for features with matching "
"descriptions are shown.")
group_ex.add_argument(
'--print-feature-string',
action='store_true',
dest='features',
help=argparse.SUPPRESS)
parser.add_argument(
'-g',
'--gui',
action='store_true',
dest='gui',
help="Open the Feature Toggler GUI")
opts = parser.parse_args(argv)
def has_opts(opt_iterable):
return any([getattr(opts, opt, None) for opt in opt_iterable])
# Check for invalid argument combinations
if not has_opts(GUI_OPTS) and not has_opts(CMDLINE_OPTS):
parser.print_help(sys.stderr)
sys.exit(1)
if has_opts(GUI_OPTS):
if has_opts(CMDLINE_OPTS):
logger.warning('WARNING: -g/--gui argument given, '
'ignoring other arguments')
for opt in CMDLINE_OPTS:
setattr(opts, opt, False)
if opts.features:
if has_opts(SET_OPTS):
parser.error('argument --print-feature-string: '
'not allowed with arguments '
'-e/--enable or -d/--disable')
if has_opts(SET_OPTS):
if opts.list:
parser.error('argument -l/--list: '
'not allowed with arguments '
'-e/--enable or -d/--disable')
enable = set()
disable = set()
if opts.enable:
enable = set(opts.enable)
if opts.disable:
disable = set(opts.disable)
enable_and_disable = enable.intersection(disable)
if enable_and_disable:
parser.error("feature{} both enabled and disabled: {}".format(
"s are" if len(enable_and_disable) > 1 else " is",
', '.join(enable_and_disable)))
opts.enable = enable
opts.disable = disable
return opts
[docs]def main(argv=None):
opts = parse_args(argv)
if opts.gui:
subprocess.call([
os.path.join(os.environ['SCHRODINGER'], 'run'),
'feature_flags_toggler_gui.py'
])
elif opts.features:
features = get_user_features()
logger.info(features)
elif opts.enable or opts.disable:
unknown = []
count = 0
total_features = len(opts.enable) + len(opts.disable)
if opts.enable:
count, enable_unknown = set_features(opts.enable, True)
unknown.extend(enable_unknown)
if opts.disable:
disable_count, disable_unknown = set_features(opts.disable, False)
count += disable_count
unknown.extend(disable_unknown)
default_count = total_features - count - len(unknown)
logger.info("Out of %d feature(s) - %d changed, %d unknown, "
"%d default state." % (total_features, count, len(unknown),
default_count))
if unknown:
logger.warning("WARNING: No such feature in default state file - "
"%s" % ' '.join(unknown))
elif opts.list:
list_features(opts.list)
warning = get_env_var_warning()
if warning:
if opts.features:
raise RuntimeError(warning)
else:
logger.warning(warning)
return
if __name__ == "__main__":
main()