Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
This utility is in charge of creating the kit and data repository for a
given RAMP event. It will also setup the database.
Parameters
----------
config : str
The path to the YAML file containing the database information.
event_config : str
The path to the YAML file containing the RAMP infomation.
setup_ramp_repo : bool, default is True
Whether or not to setup the RAMP kit and data repositories.
force : bool, default is False
Whether or not to potentially overwrite the repositories, problem and
event in the database.
"""
database_config = read_config(config, filter_section='sqlalchemy')
ramp_config = generate_ramp_config(event_config, config)
with session_scope(database_config) as session:
setup_files_extension_type(session)
if setup_ramp_repo:
setup_ramp_kit_ramp_data(
ramp_config, ramp_config['problem_name'], force
)
else:
# we do not clone the repository but we need to convert the
# notebook to html
current_directory = os.getcwd()
problem_kit_path = ramp_config['ramp_kit_dir']
os.chdir(problem_kit_path)
subprocess.check_output(["jupyter", "nbconvert", "--to", "html",
"{}_starting_kit.ipynb"
event_config : dict or str
Either the loaded configuration or the configuration YAML file. When
the configuration filename is given, ``database_config`` need to be
given as well. When a ``dict`` is provided, all paths should be given.
database_config : str, optional
The database configuration filename. It is required when
``event_config`` is a ``str``..
Returns
-------
worker_config : dict
The configuration for the RAMP worker.
"""
if isinstance(event_config, str):
ramp_config = generate_ramp_config(event_config, database_config)
event_config = read_config(
event_config, filter_section=['ramp', 'worker'])
else:
ramp_config = generate_ramp_config(event_config)
# copy the specific information for the given worker configuration
worker_config = event_config['worker'].copy()
# define the directory of the ramp-kit for the event
worker_config['kit_dir'] = ramp_config['ramp_kit_dir']
# define the directory of the ramp-data for the event
worker_config['data_dir'] = ramp_config['ramp_data_dir']
# define the directory of the submissions
worker_config['submissions_dir'] = ramp_config['ramp_submissions_dir']
# define the directory of the predictions
worker_config['predictions_dir'] = ramp_config['ramp_predictions_dir']
# define the directory of the logs
worker_config['logs_dir'] = ramp_config['ramp_logs_dir']
def generate_flask_config(config):
"""Generate the configuration to deal with Flask.
Parameters
----------
config : dict or str
Either the loaded configuration or the configuration YAML file.
Returns
-------
flask_config : dict
The configuration for the RAMP worker.
"""
if isinstance(config, str):
config = read_config(config, filter_section=['flask', 'sqlalchemy'])
flask_config = DEFAULT_CONFIG.copy()
user_flask_config = {
key.upper(): value for key, value in config['flask'].items()}
flask_config.update(user_flask_config)
database_config = config['sqlalchemy']
flask_config['SQLALCHEMY_DATABASE_URI'] = \
('{}://{}:{}@{}:{}/{}'
.format(database_config['drivername'], database_config['username'],
database_config['password'], database_config['host'],
database_config['port'], database_config['database']))
return flask_config
``event_config`` is a ``str``.
Returns
-------
ramp_config : dict
The configuration for the RAMP worker.
"""
if isinstance(event_config, str):
if (database_config is None or
not isinstance(database_config, str)):
raise ValueError(
'When "event_config" corresponds to the filename of the '
'configuration, you need to provide the filename of the '
'database as well, by assigning "database_config".'
)
config = read_config(event_config, filter_section='ramp')
path_config = os.path.dirname(
os.path.abspath(database_config)
)
else:
if 'ramp' in event_config.keys():
config = event_config['ramp']
else:
config = event_config
if not all([key in config.keys() for key in MANDATORY_DICT_PARAMS]):
raise ValueError(
'When "event_config" is a dictionary, you need to provide all '
'following keys: {}'.format(MANDATORY_DICT_PARAMS)
)
path_config = ''
ramp_config = {}