Source code for aiida.engine.processes.process_spec

# -*- coding: utf-8 -*-
###########################################################################
# Copyright (c), The AiiDA team. All rights reserved.                     #
# This file is part of the AiiDA code.                                    #
#                                                                         #
# The code is hosted on GitHub at https://github.com/aiidateam/aiida-core #
# For further information on the license, see the LICENSE.txt file        #
# For further information please visit http://www.aiida.net               #
###########################################################################
"""AiiDA specific implementation of plumpy's ProcessSpec."""
import plumpy

from .exit_code import ExitCode, ExitCodesNamespace
from .ports import InputPort, PortNamespace, CalcJobOutputPort

__all__ = ('ProcessSpec', 'CalcJobProcessSpec')


[docs]class ProcessSpec(plumpy.ProcessSpec): """Default process spec for process classes defined in `aiida-core`. This sub class defines custom classes for input ports and port namespaces. It also adds support for the definition of exit codes and retrieving them subsequently. """ METADATA_KEY = 'metadata' METADATA_OPTIONS_KEY = 'options' INPUT_PORT_TYPE = InputPort PORT_NAMESPACE_TYPE = PortNamespace
[docs] def __init__(self): super().__init__() self._exit_codes = ExitCodesNamespace()
@property def metadata_key(self): return self.METADATA_KEY @property def options_key(self): return self.METADATA_OPTIONS_KEY @property def exit_codes(self): """ Return the namespace of exit codes defined for this ProcessSpec :returns: ExitCodesNamespace of ExitCode named tuples """ return self._exit_codes
[docs] def exit_code(self, status, label, message, invalidates_cache=False): """ Add an exit code to the ProcessSpec :param status: the exit status integer :param label: a label by which the exit code can be addressed :param message: a more detailed description of the exit code :param invalidates_cache: when set to `True`, a process exiting with this exit code will not be considered for caching """ if not isinstance(status, int): raise TypeError('status should be of integer type and not of {}'.format(type(status))) if status < 0: raise ValueError('status should be a positive integer, received {}'.format(type(status))) if not isinstance(label, str): raise TypeError('label should be of str type and not of {}'.format(type(label))) if not isinstance(message, str): raise TypeError('message should be of str type and not of {}'.format(type(message))) if not isinstance(invalidates_cache, bool): raise TypeError('invalidates_cache should be of type bool and not of {}'.format(type(invalidates_cache))) self._exit_codes[label] = ExitCode(status, message, invalidates_cache=invalidates_cache)
[docs]class CalcJobProcessSpec(ProcessSpec): """Process spec intended for the `CalcJob` process class.""" OUTPUT_PORT_TYPE = CalcJobOutputPort
[docs] def __init__(self): super().__init__() self._default_output_node = None
@property def default_output_node(self): return self._default_output_node @default_output_node.setter def default_output_node(self, port_name): from aiida.orm import Dict if port_name not in self.outputs: raise ValueError('{} is not a registered output port'.format(port_name)) valid_type_port = self.outputs[port_name].valid_type valid_type_required = Dict if valid_type_port is not valid_type_required: raise ValueError( 'the valid type of a default output has to be a {} but it is {}'.format( valid_type_port, valid_type_required ) ) self._default_output_node = port_name