.rst .pdf

repository open issue suggest edit

plumpy.processes module

The main Process module

class plumpy.processes.BundleKeys

Bases: object

String keys used by the process to save its state in the state bundle.

See plumpy.processes.Process.save_instance_state() and plumpy.processes.Process.load_instance_state().

INPUTS_PARSED = 'INPUTS_PARSED'

INPUTS_RAW = 'INPUTS_RAW'

OUTPUTS = 'OUTPUTS'

class plumpy.processes.Process(*args: Any, **kwargs: Any)

Bases: plumpy.base.state_machine.StateMachine, plumpy.persistence.Savable

The Process class is the base for any unit of work in plumpy.

A process can be in one of the following states:

• CREATED

• RUNNING

• WAITING

• FINISHED

• EXCEPTED

• KILLED

as defined in the ProcessState enum.

                  ___
                 |   v
CREATED (x) --- RUNNING (x) --- FINISHED (o)
                 |   ^          /
                 v   |         /
                WAITING (x) --
                 |   ^
                  ---

* -- EXCEPTED (o)
* -- KILLED (o)

• (o): terminal state

• (x): non terminal state

When a Process enters a state is always gets a corresponding message, e.g. on entering RUNNING it will receive the on_run message. These are always called immediately after that state is entered but before being executed.

__called: bool = False

_abc_impl = <_abc_data object>

_auto_persist: Optional[Set[str]] = {'_creation_time', '_future', '_paused', '_pid', '_pre_paused_status', '_status'}

_cleanups: Optional[List[Callable[], None]]] = None

_closed = False

_create_interrupt_action(exception: plumpy.process_states.Interruption) → plumpy.futures.CancellableAction

Create an interrupt action from the corresponding interrupt exception

Parameters

exception – The interrupt exception

Returns

The interrupt action

_do_pause(state_msg: Optional[str], next_state: Optional[plumpy.process_states.State] = None) → bool

Carry out the pause procedure, optionally transitioning to the next state first

_fire_event(evt: Callable[[…], Any], *args: Any, **kwargs: Any) → None

_interrupt_action: Optional[plumpy.futures.CancellableAction] = None

_killing: Optional[plumpy.futures.CancellableAction] = None

_paused: Optional[plumpy.persistence.SavableFuture] = None

_pausing: Optional[plumpy.futures.CancellableAction] = None

_process_scope() → Generator[None, None, None]

This context manager function is used to make sure the process stack is correct meaning that globally someone can ask for Process.current() to get the last process that is on the call stack.

async _run_task(callback: Callable[[…], Any], *args: Any, **kwargs: Any) → Any

This method should be used to run all Process related functions and coroutines. If there is an exception the process will enter the EXCEPTED state.

Parameters

• callback – A function or coroutine

• args – Optional positional arguments passed to fn

• kwargs – Optional keyword arguments passed to fn

Returns

The value as returned by fn

_schedule_rpc(callback: Callable[[…], Any], *args: Any, **kwargs: Any) → concurrent.futures._base.Future

Schedule a call to a callback as a result of an RPC communication call, this will return a future that resolves to the final result (even after one or more layer of futures being returned) of the callback.

The callback will be scheduled at the working thread where the process event loop runs.

Parameters

• callback – the callback function or coroutine

• args – the positional arguments to the callback

• kwargs – the keyword arguments to the callback

Returns

a kiwi future that resolves to the outcome of the callback

_set_interrupt_action(new_action: Optional[plumpy.futures.CancellableAction]) → None

Set the interrupt action cancelling the current one if it exists :param new_action: The new interrupt action to set

_set_interrupt_action_from_exception(interrupt_exception: plumpy.process_states.Interruption) → None

Set an interrupt action from the corresponding interrupt exception

_setup_event_hooks() → None

Set the event hooks to process, when it is created or loaded(recreated).

_spec_class

alias of plumpy.process_spec.ProcessSpec

_stepping = False

add_cleanup(cleanup: Callable[], None]) → None

Add callback, which will be run when the process is being closed.

add_process_listener(listener: plumpy.process_listener.ProcessListener) → None

Add a process listener to the process.

The listener defines the actions to take when the process is triggering the specific state condition.

broadcast_receive(_comm: kiwipy.communications.Communicator, body: Any, sender: Any, subject: Any, correlation_id: Any) → Optional[concurrent.futures._base.Future]

Coroutine called when the process receives a message from the communicator

Parameters

• _comm – the communicator that sent the message

• msg – the message

call_soon(callback: Callable[[…], Any], *args: Any, **kwargs: Any) → plumpy.events.ProcessCallback

Schedule a callback to what is considered an internal process function (this needn’t be a method). If it raises an exception it will cause the process to fail.

callback_excepted(_callback: Callable[[…], Any], exception: Optional[BaseException], trace: Optional[traceback]) → None

close() → None

Calling this method indicates that this process should not ran anymore and will trigger any runtime resources (such as the communicator connection) to be cleaned up. The state of the process will still be accessible.

It is safe to call this method multiple times.

create_initial_state() → plumpy.process_states.State

This method is here to override its superclass.

Automatically enter the CREATED state when the process is created.

Returns

A Created state

property creation_time

The creation time of this Process as returned by time.time() when instantiated :return: The creation time

classmethod current() → Optional[plumpy.processes.Process]

Get the currently running process i.e. the one at the top of the stack

Returns

the currently running process

decode_input_args(encoded: Any) → Any

Decode saved input arguments as they came from the saved instance state plumpy.persistence.Bundle. The decoded inputs should contain no reference to the encoded inputs that were passed in. This often will mean making a deepcopy of the encoded input dictionary.

Parameters

encoded –

Returns

The decoded input args

classmethod define(_spec: plumpy.process_spec.ProcessSpec) → None

Define the specification of the process.

Normally should be overridden by subclasses.

done() → bool

Return True if the call was successfully killed or finished running.

Deprecated since version 0.18.6: Use the has_terminated method instead

encode_input_args(inputs: Any) → Any

Encode input arguments such that they may be saved in a plumpy.persistence.Bundle. The encoded inputs should contain no reference to the inputs that were passed in. This often will mean making a deepcopy of the input dictionary.

Parameters

inputs – A mapping of the inputs as passed to the process

Returns

The encoded inputs

exception() → Optional[BaseException]

Return exception, if the process is terminated in excepted state.

execute() → Optional[Dict[str, Any]]

Execute the process. This will return if the process terminates or is paused.

Returns

None if not terminated, otherwise self.outputs

fail(exception: Optional[BaseException], trace_back: Optional[traceback]) → None

Fail the process in response to an exception :param exception: The exception that caused the failure :param trace_back: Optional exception traceback

future() → plumpy.persistence.SavableFuture

Return a savable future representing an eventual result of an asynchronous operation.

The result is set at the terminal state.

classmethod get_description() → Dict[str, Any]

Get a human readable description of what this Process does.

Returns

The description.

classmethod get_name() → str

Return the process class name.

classmethod get_state_classes() → Dict[Hashable, Type[plumpy.process_states.State]]

classmethod get_states() → Sequence[Type[plumpy.process_states.State]]

Return all allowed states of the process.

get_status_info(out_status_info: dict) → None

Return updated status information of process.

Parameters

out_status_info – the old status

has_terminated() → bool

Return whether the process was terminated.

init(*args: Any, **kwargs: Any) → None

property inputs

Return the parsed inputs.

property is_killing

Return if the process is already being killed.

property is_successful

Return whether the result of the process is considered successful.

Returns

boolean, True if the process is in Finished state with successful attribute set to True

kill(msg: Optional[str] = None) → Union[bool, _asyncio.Future]

Kill the process :param msg: An optional kill message

killed() → bool

Return whether the process is killed.

killed_msg() → Optional[str]

Return the killed message.

launch(process_class: Type[Process], inputs: Optional[dict] = None, pid: Optional[Hashable] = None, logger: Optional[logging.Logger] = None) → plumpy.processes.Process

Start running the nested process.

The process is started asynchronously, without blocking other task in the event loop.

load_instance_state(saved_state: MutableMapping[str, Any], load_context: plumpy.persistence.LoadSaveContext) → None

Load the process from its saved instance state.

Parameters

• saved_state – A bundle to load the state from

• load_context – The load context

log_with_pid(level: int, msg: str) → None

Log the message with the process pid.

property logger

Return the logger for this class.

If not set, return the default logger.

Returns

The logger.

property loop

Return the event loop of the process.

message_receive(_comm: kiwipy.communications.Communicator, msg: Dict[str, Any]) → Any

Coroutine called when the process receives a message from the communicator

Parameters

• _comm – the communicator that sent the message

• msg – the message

Returns

the outcome of processing the message, the return value will be sent back as a response to the sender

on_close(*args: Any, **kwargs: Any) → None

on_create(*args: Any, **kwargs: Any) → None

on_entered(from_state: Optional[plumpy.process_states.State]) → None

on_entering(state: plumpy.process_states.State) → None

on_except(*args: Any, **kwargs: Any) → None

on_excepted(*args: Any, **kwargs: Any) → None

on_exit_running(*args: Any, **kwargs: Any) → None

on_exit_waiting(*args: Any, **kwargs: Any) → None

on_exiting() → None

on_finish(*args: Any, **kwargs: Any) → None

on_finished(*args: Any, **kwargs: Any) → None

on_kill(*args: Any, **kwargs: Any) → None

on_killed(*args: Any, **kwargs: Any) → None

on_output_emitted(output_port: str, value: Any, dynamic: bool) → None

on_output_emitting(output_port: str, value: Any) → None

Output is about to be emitted.

on_paused(*args: Any, **kwargs: Any) → None

on_pausing(*args: Any, **kwargs: Any) → None

on_playing(*args: Any, **kwargs: Any) → None

on_run(*args: Any, **kwargs: Any) → None

on_running(*args: Any, **kwargs: Any) → None

on_terminated() → None

Call when a terminal state is reached.

on_wait(*args: Any, **kwargs: Any) → None

on_waiting(*args: Any, **kwargs: Any) → None

out(output_port: str, value: Any) → None

Record an output value for a specific output port. If the output port matches an explicitly defined Port it will be validated against that. If not it will be validated against the PortNamespace, which means it will be checked for dynamicity and whether the type of the value is valid

Parameters

• output_port – the name of the output port, can be namespaced

• value – the value for the output port

Raises

ValueError if the output value is not validated against the port

property outputs

Get the current outputs emitted by the Process. These may grow over time as the process runs.

Returns

A mapping of {output_port: value} outputs

pause(msg: Optional[str] = None) → Union[bool, plumpy.futures.CancellableAction]

Pause the process.

Parameters

msg – an optional message to set as the status. The current status will be saved in the private _pre_paused_status attribute, such that it can be restored when the process is played again.

Returns

False if process is already terminated, True if already paused or pausing, a CancellableAction to pause if the process was running steps

property paused

Return whether the process was being paused.

property pid

Return the pid of the process.

play() → bool

Play a process. Returns True if after this call the process is playing, False otherwise

Returns

True if playing, False otherwise

property raw_inputs

The AttributesFrozendict of inputs (if not None).

classmethod recreate_from(saved_state: MutableMapping[str, Any], load_context: Optional[plumpy.persistence.LoadSaveContext] = None) → plumpy.processes.Process

Recreate a process from a saved state, passing any positional and keyword arguments on to load_instance_state

Parameters

• saved_state – The saved state to load from

• load_context – The load context to use

Returns

An instance of the object with its state loaded from the save state.

recreate_state(saved_state: plumpy.persistence.Bundle) → plumpy.process_states.State

Create a state object from a saved state

Parameters

saved_state – The saved state

Returns

An instance of the object with its state loaded from the save state.

remove_process_listener(listener: plumpy.process_listener.ProcessListener) → None

Remove a process listener from the process.

result() → Any

Get the result from the process if it is finished. If the process was killed then a KilledError will be raise. If the process has excepted then the failing exception will be raised. If in any other state this will raise an InvalidStateError. :return: The result of the process

resume(*args: Any) → None

Start running the process again.

run() → Any

This function will be run when the process is triggered. It should be overridden by a subclass.

save_instance_state(out_state: MutableMapping[str, Any], save_context: Optional[plumpy.persistence.LoadSaveContext]) → None

Ask the process to save its current instance state.

Parameters

• out_state – A bundle to save the state to

• save_context – The save context

set_logger(logger: logging.Logger) → None

Set the logger of the process.

set_status(status: Optional[str]) → None

Set the status message of the process.

classmethod spec() → plumpy.process_spec.ProcessSpec

property status

Return the status massage of the process.

step() → None

Run a step.

The step is run synchronously with steps in its own process, and asynchronously with steps in other processes.

The execute function running in this method is dependent on the state of the process.

async step_until_terminated() → None

If the process has not terminated, run the current step and wait until the step finished.

This is the function run by the event loop (not step).

successful() → bool

Returns whether the result of the process is considered successful Will raise if the process is not in the FINISHED state

transition_excepted(_initial_state: Any, final_state: plumpy.process_states.ProcessState, exception: Exception, trace: traceback) → None

property uuid

Return the UUID of the process

class plumpy.processes.ProcessSpec

Bases: object

A class that defines the specifications of a plumpy.Process, this includes what its inputs, outputs, etc are.

All methods to modify the spec should have declarative names describe the spec e.g.: input, output

Every Process class has one of these.

INPUT_PORT_TYPE

alias of plumpy.ports.InputPort

NAME_INPUTS_PORT_NAMESPACE: str = 'inputs'

NAME_OUTPUTS_PORT_NAMESPACE: str = 'outputs'

OUTPUT_PORT_TYPE

alias of plumpy.ports.OutputPort

PORT_NAMESPACE_TYPE

alias of plumpy.ports.PortNamespace

_create_port(port_namespace: plumpy.ports.PortNamespace, port_class: Type[Union[plumpy.ports.Port, plumpy.ports.PortNamespace]], name: str, **kwargs: Any) → None

Create a new Port of a given class and name in a given PortNamespace

Parameters

• port_namespace – PortNamespace to which to add the port

• port_class – class of the Port to create

• name – name of the port to create

• kwargs – options for the port

static _expose_ports(process_class: Type[Process], source: plumpy.ports.PortNamespace, destination: plumpy.ports.PortNamespace, expose_memory: Dict[Optional[str], Dict[Type[Process], Sequence[str]]], namespace: Optional[str], exclude: Optional[Sequence[str]], include: Optional[Sequence[str]], namespace_options: Optional[dict] = None) → None

Expose ports from a source PortNamespace of the ProcessSpec of a Process class into the destination PortNamespace of this ProcessSpec. If the namespace is specified, the ports will be exposed in that sub namespace. The set of ports can be restricted using the mutually exclusive exclude and include tuples. The namespace_options will be used to override the properties of the PortNamespace into which the ports are exposed, whether that has been newly created or existed already.

Parameters

• process_class – the Process class whose outputs to expose

• source – the PortNamespace whose ports are to be exposed

• destination – the PortNamespace into which the ports are to be exposed

• namespace – a namespace in which to place PortNamespace with the exposed outputs

• exclude – input ports that are to be excluded

• include – input ports that are to be included

• namespace_options – a dictionary with mutable PortNamespace property values to override

expose_inputs(process_class: Type[Process], namespace: Optional[str] = None, exclude: Optional[Sequence[str]] = None, include: Optional[Sequence[str]] = None, namespace_options: Optional[dict] = None) → None

This method allows one to automatically add the inputs from another Process to this ProcessSpec. The optional namespace argument can be used to group the exposed inputs in a separated PortNamespace. The exclude and include arguments can be used to restrict the set of ports that are exposed. Note that these two options are mutually exclusive.

Parameters

• process_class – the Process class whose inputs to expose

• namespace – a namespace in which to place the exposed inputs

• exclude – input ports that are to be excluded

• include – input ports that are to be included

• namespace_options – a dictionary with mutable PortNamespace property values to override

expose_outputs(process_class: Type[Process], namespace: Optional[str] = None, exclude: Optional[Sequence[str]] = None, include: Optional[Sequence[str]] = None, namespace_options: Optional[dict] = None) → None

This method allows one to automatically add the ouputs from another Process to this ProcessSpec. The optional namespace argument can be used to group the exposed outputs in a separated PortNamespace. The exclude and include arguments can be used to restrict the set of ports that are exposed. Note that these two options are mutually exclusive.

Parameters

• process_class – the Process class whose outputs to expose

• namespace – a namespace in which to place the exposed outputs

• exclude – input ports that are to be excluded

• include – input ports that are to be included

• namespace_options – a dictionary with mutable PortNamespace property values to override

get_description() → Dict[str, Any]

Get a description of this process specification

Returns

a dictionary with the descriptions of the input and output port namespaces

has_input(name: str) → bool

Return whether the input port namespace contains a port with the given name

Parameters

name – key of the port in the input port namespace

has_output(name: str) → bool

Return whether the output port namespace contains a port with the given name

Parameters

name – key of the port in the output port namespace

input(name: str, **kwargs: Any) → None

Define an input port in the input port namespace

Parameters

• name – name of the input port to create

• kwargs – options for the input port

input_namespace(name: str, **kwargs: Any) → None

Create a new PortNamespace in the input port namespace. The keyword arguments will be passed to the PortNamespace constructor. Any intermediate port namespaces that need to be created for a nested namespace, will take constructor defaults

Parameters

• name – namespace of the new port namespace

• kwargs – keyword arguments for the PortNamespace constructor

property inputs

Get the input port namespace of the process specification

Returns

the input PortNamespace

property logger

property namespace_separator

output(name: str, **kwargs: Any) → None

Define an output port in the output port namespace

Parameters

• name – name of the output port to create

• kwargs – options for the output port

output_namespace(name: str, **kwargs: Any) → None

Create a new PortNamespace in the output port namespace. The keyword arguments will be passed to the PortNamespace constructor. Any intermediate port namespaces that need to be created for a nested namespace, will take constructor defaults

Parameters

• name – namespace of the new port namespace

• kwargs – keyword arguments for the PortNamespace constructor

property outputs

Get the output port namespace of the process specification

Returns

the outputs PortNamespace

property ports

seal() → None

Seal this specification disallowing any further changes

property sealed

Indicates if the spec is sealed or not

Returns

True if sealed, False otherwise

Return type

bool

exception plumpy.processes.TransitionFailed(initial_state: plumpy.base.state_machine.State, final_state: Optional[State] = None, traceback_str: Optional[str] = None)

Bases: Exception

A state transition failed

_format_msg() → str

plumpy.process_states module plumpy.settings module