%env CONSOLE_WIDTH=140
from kiara.interfaces.python_api.workflow import Workflow
from kiara.utils.jupyter import graph_to_image
from kiara.utils.cli import terminal_print_model
doc = """Returns 'True' if exactly one of it's two inputs is 'True'."""
workflow = Workflow.create("logic.xor", doc=doc, replace_existing_alias=True)
# Creating step: or
workflow.add_step(operation="logic.or", step_id="or")
# Creating step: nand
step_nand_config = {'constants': {}, 'defaults': {}, 'pipeline_name': 'logic.nand', 'steps': [{'module_type': 'logic.and', 'module_config': {}, 'is_resolved': False, 'step_id': 'and', 'manifest_src': {'module_type': 'logic.and', 'module_config': {}, 'is_resolved': False}, 'input_links': {}, 'module_details': {'python_class_name': 'AndModule', 'python_module_name': 'kiara_plugin.core_types.modules.boolean', 'full_name': 'kiara_plugin.core_types.modules.boolean.AndModule', 'module_config': {'constants': {}, 'defaults': {}, 'delay': 0.0}, 'inputs_schema': {'a': {'type': 'boolean', 'type_config': {}, 'default': '__not_set__', 'optional': False, 'is_constant': False, 'doc': {'description': 'A boolean describing this input state.', 'doc': None}}, 'b': {'type': 'boolean', 'type_config': {}, 'default': '__not_set__', 'optional': False, 'is_constant': False, 'doc': {'description': 'A boolean describing this input state.', 'doc': None}}}, 'outputs_schema': {'y': {'type': 'boolean', 'type_config': {}, 'default': '__not_set__', 'optional': False, 'is_constant': False, 'doc': {'description': 'A boolean describing the module output state.', 'doc': None}}}}, 'doc': {'description': '-- n/a --', 'doc': None}}, {'module_type': 'logic.not', 'module_config': {}, 'is_resolved': False, 'step_id': 'not', 'manifest_src': {'module_type': 'logic.not', 'module_config': {}, 'is_resolved': False}, 'input_links': {'a': [{'step_id': 'and', 'value_name': 'y', 'sub_value': None}]}, 'module_details': {'python_class_name': 'NotModule', 'python_module_name': 'kiara_plugin.core_types.modules.boolean', 'full_name': 'kiara_plugin.core_types.modules.boolean.NotModule', 'module_config': {'constants': {}, 'defaults': {}, 'delay': 0.0}, 'inputs_schema': {'a': {'type': 'boolean', 'type_config': {}, 'default': '__not_set__', 'optional': False, 'is_constant': False, 'doc': {'description': 'A boolean describing this input state.', 'doc': None}}}, 'outputs_schema': {'y': {'type': 'boolean', 'type_config': {}, 'default': '__not_set__', 'optional': False, 'is_constant': False, 'doc': {'description': 'A boolean describing the module output state.', 'doc': None}}}}, 'doc': {'description': '-- n/a --', 'doc': None}}], 'input_aliases': {'and.a': 'a', 'and.b': 'b'}, 'output_aliases': {'not.y': 'y'}, 'doc': {'description': "Returns 'False' if both inputs are 'True'.", 'doc': None}, 'context': {}}
workflow.add_step(
operation="pipeline",
module_config=step_nand_config,
step_id="nand")
# Creating step: and
workflow.add_step(operation="logic.and", step_id="and")
# Connecting input(s) of step 'and'
workflow.connect_fields("and.a", "or.y")
workflow.connect_fields("and.b", "nand.y")
Setting workflow input/output names (optional)¶
To make our workflow nicer to use, we can set aliases for its inputs and outputs.
workflow.set_input_alias(input_field="or.a", alias="a")
workflow.set_input_alias(input_field="or.b", alias="b")
workflow.set_input_alias(input_field="nand.a", alias="a")
workflow.set_input_alias(input_field="nand.b", alias="b")
workflow.set_output_alias(output_field="and.y", alias="y")
Workflow information ¶
After our workflow is wired up, we look can look at its structure, and other properties.
Workflow status¶
A workflow consists of a series of 'states', the most relevant is always the most recent one. We can investigate that latest states details like so:
workflow.current_state
Pipeline execution graph¶
Let's look at the current execution graph for the current workflow pipeline:
graph_to_image(workflow.pipeline.execution_graph)
Workflow inputs ¶
Once a workflow has an assembled pipeline, we can set it's inputs. We use the input field
names that we got from the result of the workflow.current_state call.
workflow.set_input("a", True)
workflow.set_input("b", True)
# process all workflow steps that can be processed
workflow.process_steps()
# print the current state, after we set our inputs
workflow.current_state
Workflow outputs ¶
To print the actual data of the workflows' current outputs, we call the current_output_values property of the workflow object:
workflow.current_output_values
Workflow snapshot ¶
So far, our workflow only exists in memory. If we want to save it so we can have a look at it again at a later stage, we can snapshot the current state, which will save the current structure of the internal pipeline, as well as all inputs that are currently used. In addition, this will register the workflow under the alias we specified on top of this file when creating the Workflow object (in our case: logic.xor).
If we would not not specify save=True, the structure of the pipeline and inputs would still be frozen and kept, but only in memory, and we'd only be able to access it in our current session.
workflow.snapshot(save=True)
Now, we can access our workflow in other environments, for example from the commandline:
! kiara workflow list
! kiara workflow explain logic.xor