#
# Command-line interface script to feature flags
#
import argparse
import os
import re
import sys
from typing import Set
import pymmlibs
pymmlibs.mmerr_set_mmlibs()
# isort: split
from schrodinger.utils import log
from schrodinger.utils import mmutil
from schrodinger.utils import subprocess
from .write import read_json_data
from .write import write_features_json
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)
ENV_FEATURE_ENABLED, ENV_FEATURE_DISABLED, ENV_FEATURE_SEP = '+', '-', ' '
[docs]def get_features():
"""
Get the feature flags from site and user state file based
on the existence.
:rtype: tuple
:return: tuple of feature dictionaries - site flags, user flags and combined flags
"""
combined_feature_flags = {}
site_feature_file = mmutil.get_feature_flags_site_state_file()
if site_feature_file:
combined_feature_flags = read_json_data(site_feature_file)
site_features = set(combined_feature_flags)
else:
site_features = set()
user_feature_flags = read_json_data(
mmutil.get_feature_flags_user_state_file())
user_features = set(user_feature_flags)
site_features = site_features.difference(user_features)
for k in user_feature_flags:
if k not in combined_feature_flags:
combined_feature_flags[k] = user_feature_flags[k]
return site_features, user_features, combined_feature_flags
[docs]def set_features(features, desired_state):
"""
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.
"""
# feature flags persisted in ~/.schrodinger/featureflags.json
persisted_feature_flags = 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)
if default_state == mmutil.FEATURE_NOT_PRESENT:
not_available_features.append(feature)
continue
# Same as the state that is in site config file, don't set in user file
user_state = mmutil.feature_flag_user_state_s(feature)
if user_state == desired_state:
continue
# Flag has been persisted and new state identical to default_state
if default_state == desired_state and user_state == mmutil.FEATURE_NOT_PRESENT:
continue
# Update existing feature (will overwrite existing `feature`, if any)
persisted_feature_flags[feature] = {
"Feature": feature,
"Enabled": desired_state
}
count = count + 1
if count:
feature_flags_format = list(persisted_feature_flags.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_nondefault_features() -> str:
"""
Return a string usable as the SCHRODINGER_FEATURE_FLAGS environment variable
that includes any non-default settings in either the user settings or site
settings, and any user settings that restore non-default site settings to
their defaults.
"""
site_nondefaults = get_site_nondefaults()
total_nondefaults = set(mmutil.feature_flags_get_nondefault_map())
# Take the union of both site and total nondefaults.
# Rationale:
# - If something is in site_nondefaults but not in total_nondefaults, user
# state must be overriding it back to default state and needs to be kept.
# - If something is in total_nondefaults but not in site_nondefault, user
# state must be overriding it to a nondefault state and needs to be kept.
modified_features = []
for feature in site_nondefaults | total_nondefaults:
enabled = mmutil.feature_flag_is_enabled_s(feature)
if enabled:
modified_features.append(ENV_FEATURE_ENABLED + feature)
else:
modified_features.append(ENV_FEATURE_DISABLED + feature)
return ENV_FEATURE_SEP.join(modified_features)
[docs]def get_site_nondefaults() -> Set[str]:
"""
Return the nondefault feature flags in the site state file.
"""
site_feature_file = mmutil.get_feature_flags_site_state_file()
if not site_feature_file:
return set()
try:
site_fflags = read_json_data(site_feature_file)
except ValueError:
return set()
site_nondefaults = set()
for feature in site_fflags.values():
feature_s = feature["Feature"]
enabled = feature["Enabled"]
default_state = mmutil.feature_flag_default_state_s(feature_s)
if default_state == mmutil.FEATURE_NOT_PRESENT:
continue
if default_state != enabled:
site_nondefaults.add(feature_s)
return site_nondefaults
[docs]def get_user_feature_env(feature_const: int, enable: bool = True):
"""
Enable a feature flag and return dict with custom feature flags.
:param int feature_const: Feature to be enabled/disabled
:param bool enable: If True enable, if False disable
:raise ValueError: If feature is unknown
:rtype: dict
:return: Dict with key SCHRODINGER_FEATURE_FLAGS and all the custom flags as
value
"""
feature = mmutil.feature_flags_get_feature_string(feature_const)
if not feature:
raise ValueError('Unknown feature: %s.' % feature_const)
user_features = get_nondefault_features().split(ENV_FEATURE_SEP)
# Try to remove feature if it is already present in the user features
for state in (ENV_FEATURE_ENABLED, ENV_FEATURE_DISABLED):
try:
user_features.remove(state + feature)
except ValueError:
pass
state = ENV_FEATURE_ENABLED if enable else ENV_FEATURE_DISABLED
user_features.append(state + feature)
features_str = ENV_FEATURE_SEP.join(user_features)
return {SCHRODINGER_FEATURE_FLAGS: features_str}
[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_nondefault_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()