Source code for schrodinger.application.jaguar.workflow_validation
"""
Workflow keywords input validation and specialized Exceptions
"""
# Contributors: Mark A. Watson
import re
import schrodinger.application.jaguar.utils as utils
from schrodinger.application.jaguar.basis import num_functions_all_atoms
from schrodinger.application.jaguar.exceptions import JaguarUnsupportedBasisSet
from schrodinger.application.jaguar.exceptions import JaguarUserFacingException
#------------------------------------------------------------------------------
[docs]class WorkflowKeywordException(JaguarUserFacingException):
    """
    Base exception class for all custom Workflow keyword validation errors
    """ 
[docs]class WorkflowConservationError(JaguarUserFacingException):
    """
    Runtime error due to a failure to conserve something
    """ 
[docs]class WorkflowKeywordError(WorkflowKeywordException):
    """
    Exception class raised when nonexistant Workflow keyword is requested
    """
[docs]    def __init__(self, keyword, allowed_keywords):
        """
        :type  keyword: string
        :param keyword: input keyword
        :type  allowed_keywords: list
        :param allowed_keywords: list of allowed keywords
        """
        msg = "'%s' is not a keyword in this workflow." % keyword
        super().__init__(msg)
        self.keyword = keyword
        self.allowed_keywords = allowed_keywords  
[docs]class WorkflowKeywordValueTypeError(WorkflowKeywordException):
    """
    Exception class raised when Workflow keyword value has wrong type
    """
[docs]    def __init__(self, keyword, value, valid_type):
        """
        :type  keyword: string
        :param keyword: input keyword
        :type  value: depends on keyword
        :param value: input value
        :type  valid_type: python type
        :param valid_type: types as documented in the appropriate
            `*_keywords.py` file
        """
        msg = "%s of %s is not a valid type for keyword '%s'; expected %s." % \
            
(value, type(value), keyword, valid_type)
        super().__init__(msg)
        self.keyword = keyword
        self.value = value
        self.valid_type = valid_type  
[docs]class WorkflowKeywordValueError(WorkflowKeywordException):
    """
    Exception class raised when Workflow keyword value is invalid
    """
[docs]    def __init__(self, keyword, value, choices):
        """
        :type  keyword: string
        :param keyword: input keyword
        :type  value: depends on keyword
        :param value: input value
        :type  choices: list
        :param choices: valid choices associated with a keyword
        """
        self.keyword = keyword
        self.value = value
        self.allowed_values = choices
        msg = "%s is not an allowed value for keyword '%s'; allowed values are %s" % (
            value, keyword, str(choices))
        super().__init__(msg)  
[docs]class WorkflowKeywordConflictError(WorkflowKeywordException):
    """
    Exception class raised when Workflow keywords have conflicting values
    """
[docs]    def __init__(self, mykey, key, value):
        """
        :type  mykey: string
        :param mykey: keyword name
        :type  key: string
        :param key: required keyword name
        :type  value: any
        :param value: required keyword value
        """
        msg = "'%s' requires '%s' to have value '%s'" % (mykey, key, value)
        super().__init__(msg)
        self.keyword = mykey
        self.required_key = key
        self.required_value = value  
[docs]class JaguarKeywordConflict(WorkflowKeywordException):
    """
    Exception class raised when a Jaguar keyword is set that
    we wish to prevent in this workflow
    """ 
SCHEMA_ERRORS = {
    re.compile("expected.*for dictionary value.*"): WorkflowKeywordValueTypeError,
    re.compile("not a valid value for.*"): WorkflowKeywordValueError,
    re.compile("invalid list value.*"): WorkflowKeywordValueError
}
[docs]def raise_voluptuous_exception(exception, kwd):
    """
    Re-raise voluptuous Exceptions as WorkflowKeywordException's
    """
    exps = [v for k, v in SCHEMA_ERRORS.items() if k.match(str(exception))]
    if exps:
        raise exps[0](kwd.name, kwd.value, kwd.valid_type)
    else:
        raise exception 
def _calculate_num_electrons(structures):
    """
    Return the total number of electrons in structure(s) considering charges.
    :param structures: structures whose electrons must be added up
    :type structures: Structure object or iterable of Structure objects
    :return: the total number of electrons
    :rtype: int
    """
    if hasattr(structures, "__iter__"):
        n_elecs = sum([utils.get_number_electrons(st) for st in structures])
        charge = sum([utils.get_total_charge(st) for st in structures])
    else:
        n_elecs = utils.get_number_electrons(structures)
        charge = utils.get_total_charge(structures)
    return n_elecs - charge
[docs]def estate_is_physical(strs, charge, mult):
    """
    Check whether or not the requested electronic state is plausible.
    This is done by ensuring the number of electrons is consistent with the
    requested charge/multiplicity.  Raises a WorkflowConservationError
    :type strs: Structure object or iterable of Structure objects
    :param strs: the reactants or reactant complex
    :type charge: int
    :param charge: overall charge
    :type mult: int
    :type mult: overall spin multiplicity
    """
    if hasattr(strs, "__iter__"):
        n_elecs = sum([utils.get_number_electrons(st) for st in strs])
    else:
        n_elecs = utils.get_number_electrons(strs)
    # multiplicity - 1 = number of unpaired electrons
    ok = ((n_elecs - charge) - (mult - 1)) % 2 == 0
    if not ok:
        msg = f"Charge/Multiplicity specification (charge = {charge}, multiplicity = {mult}) is not consistent with the number of electrons"
        raise WorkflowConservationError(msg) 
[docs]def charge_is_consistent(strs, charge):
    """
    Tests that the sum of molecular charges is consistent with the total charge.
    raises WorkflowConservationError if this criterion is not satisfied.
    :type strs: Structure or iterable of Structure objects
    :param strs: reactant or product structure(s) to check
    :type charge: int
    :param charge: overall charge of reaction
    """
    if hasattr(strs, "__iter__"):
        q = sum(map(utils.get_total_charge, strs))
    else:
        q = utils.get_total_charge(strs)
    if q != charge:
        q_str = str(q)
        charge_str = str(charge)
        if q > 0:
            q_str = '+' + q_str
        if charge > 0:
            charge_str = '+' + charge_str
        msg = f"Total charge specified ({charge_str}) is inconsistent with the charge of molecules ({q_str})"
        raise WorkflowConservationError(msg) 
[docs]def basis_set_is_valid(strs, basis):
    """
    Checks that the given basis set is defined for all
    atoms in the structures.  A JaguarUnsupportedBasisSet is
    raised if the basis is not supported.
    :type strs: list
    :param strs: list of structures to check
    :type basis: string
    :param basis: name of basis set
    """
    for st in strs:
        nbasis, ps_supported, basis_per_atom = num_functions_all_atoms(
            basis, st)
        if nbasis == 0:
            raise JaguarUnsupportedBasisSet(
                "Chosen basis set (%s) is not supported for all of the atom types in the input structures."
                % basis)