Source code for pyretis.inout.checker
# -*- coding: utf-8 -*-
# Copyright (c) 2023, PyRETIS Development Team.
# Distributed under the LGPLv2.1+ License. See LICENSE for more info.
"""This module checks that the inputs are meaningful.
Main methods defined here
~~~~~~~~~~~~~~~~~~~~~~~~~
check_ensemble (:py:func:`.check_ensemble`)
Function to check the ensemble settings.
check_interfaces (:py:func:`.check_interfaces`)
Function to check engine set-up.
check_for_bullshitt (:py:func:`.check_for_bullshitt`)
Function to compare nested dicts and lists.
check_engine (:py:func:`.check_engine`)
Function to check engine set-up.
"""
import logging
logger = logging.getLogger(__name__) # pylint: disable=C0103
logger.addHandler(logging.NullHandler())
__all__ = ['check_interfaces', 'check_for_bullshitt',
'check_engine', 'check_ensemble']
ENSEMBLE_TASKS = ['retis', 'tis', 'repptis', 'pptis']
[docs]def check_ensemble(settings):
"""Check that the ensemble input parameters are complete.
Parameters
----------
settings : dict
The settings needed to set up the simulation.
"""
if 'ensemble' in settings:
savelambda = []
for ens in settings['ensemble']:
if 'interface' in ens:
savelambda.append(ens['interface'])
if not ens['interface'] \
in settings['simulation']['interfaces']:
msg = f"The ensemble interface {ens['interface']} "\
"is not present in the simulation interface list"
break
else:
msg = "An ensemble is present without reference interface"
break
if not is_sorted(savelambda):
msg = "Interface positions in the ensemble simulation "\
"are NOT properly sorted (ascending order)"
else:
msg = "No ensemble in settings"
if 'msg' in locals():
raise ValueError(msg)
return True
[docs]def check_engine(settings):
"""Check the engine settings.
Checks that the input engine settings are correct, and
automatically determine the 'internal' or 'external'
engine setting.
Parameters
----------
settings : dict
The current input settings.
"""
msg = []
if 'engine' not in settings:
msg += ['The section engine is missing']
elif settings['engine'].get('type') == 'external':
if 'input_path' not in settings['engine']:
msg += ['The section engine requires an input_path entry']
if 'gmx' in settings['engine'] and \
'gmx_format' not in settings['engine']:
msg += ['File format is not specified for the engine']
elif 'cp2k' in settings['engine'] and \
'cp2k_format' not in settings['engine']:
msg += ['File format is not specified for the engine']
if msg:
msgtxt = '\n'.join(msg)
logger.critical(msgtxt)
return False
return True
[docs]def check_for_bullshitt(settings):
"""Do what is stated.
Just for the input settings.
Parameters
----------
settings : dict
The current input settings.
"""
if (settings['simulation']['task'] in ENSEMBLE_TASKS and
len(settings['simulation']['interfaces']) < 3):
msg = "Insufficient number of interfaces for "\
f"{settings['simulation']['task']}"
elif settings['simulation']['task'] in ENSEMBLE_TASKS:
if not is_sorted(settings['simulation']['interfaces']):
msg = "Interface lambda positions in the simulation "\
"entry are NOT sorted (small to large)"
if 'ensemble' in settings:
savelambda = []
for i_ens, ens in enumerate(settings['ensemble']):
if 'ensemble_number' not in ens and \
'interface' not in ens:
msg = "An ensemble has been introduced without "\
"references (interface in ensemble settings)"
else:
savelambda.append(settings['simulation']['interfaces']
[i_ens])
if 'interface' in ens and ens['interface'] \
not in settings['simulation']['interfaces']:
msg = "An ensemble has been introduced with an "\
"interface not listed in the simulation interfaces"
if 'msg' in locals():
raise ValueError(msg)
[docs]def check_interfaces(settings):
"""Check that the interfaces are properly defined.
Parameters
----------
settings : dict
The current input settings.
"""
msg = []
if settings['simulation'].get('flux', False) and \
not settings['simulation']['zero_ensemble']:
msg += ['Settings for flux and zero_ensemble are inconsistent.']
if settings['simulation']['task'] in ENSEMBLE_TASKS:
if len(settings['simulation']['interfaces']) < 3:
msg += ['Insufficient number of interfaces for {}'
.format(settings['simulation']['task'])]
if not is_sorted(settings['simulation']['interfaces']):
msg += ['Interface positions in the simulation interfaces ']
msg += ['input are NOT properly sorted (ascending order)']
if msg:
msgtxt = '\n'.join(msg)
logger.critical(msgtxt)
return False
return True
def is_sorted(lll):
"""Check if a list is sorted."""
return all(aaa <= bbb for aaa, bbb in zip(lll[:-1], lll[1:]))