Skip to content

pipeline

Classes

PipelineTransformer

Bases: SourceTransformer

Source code in kiara/renderers/included_renderers/pipeline.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class PipelineTransformer(SourceTransformer):
    def __init__(self, kiara: "Kiara"):
        self._kiara: "Kiara" = kiara
        super().__init__()

    def retrieve_supported_python_classes(self) -> Iterable[Type]:
        return [PipelineConfig, Pipeline, PipelineStructure, str, Mapping]

    def retrieve_supported_inputs_descs(self) -> Union[str, Iterable[str]]:
        return [
            "a registered pipeline operation",
            "a path to a pipeline file",
            "the pipeline configuration as a Python dict",
        ]

    def validate_and_transform(self, source: Any) -> Union[Pipeline, None]:

        if isinstance(source, Pipeline):
            return source
        elif isinstance(source, (PipelineConfig, PipelineStructure, Mapping, str)):
            pipeline = Pipeline.create_pipeline(kiara=self._kiara, pipeline=source)
            return pipeline
        else:
            return None

Functions

retrieve_supported_python_classes() -> Iterable[Type]
Source code in kiara/renderers/included_renderers/pipeline.py 
28
29
def retrieve_supported_python_classes(self) -> Iterable[Type]:
    return [PipelineConfig, Pipeline, PipelineStructure, str, Mapping]
retrieve_supported_inputs_descs() -> Union[str, Iterable[str]]
Source code in kiara/renderers/included_renderers/pipeline.py 
31
32
33
34
35
36
def retrieve_supported_inputs_descs(self) -> Union[str, Iterable[str]]:
    return [
        "a registered pipeline operation",
        "a path to a pipeline file",
        "the pipeline configuration as a Python dict",
    ]
validate_and_transform(source: Any) -> Union[Pipeline, None]
Source code in kiara/renderers/included_renderers/pipeline.py 
38
39
40
41
42
43
44
45
46
def validate_and_transform(self, source: Any) -> Union[Pipeline, None]:

    if isinstance(source, Pipeline):
        return source
    elif isinstance(source, (PipelineConfig, PipelineStructure, Mapping, str)):
        pipeline = Pipeline.create_pipeline(kiara=self._kiara, pipeline=source)
        return pipeline
    else:
        return None

PipelineRendererHtml

Bases: BaseJinjaRenderer[Type[Pipeline], RenderInputsSchema]

Renders a pipeline structure as static html page.

Source code in kiara/renderers/included_renderers/pipeline.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class PipelineRendererHtml(BaseJinjaRenderer[Type[Pipeline], RenderInputsSchema]):
    """Renders a pipeline structure as static html page."""

    _renderer_name = "pipeline_html"

    # _render_profiles: Mapping[str, Mapping[str, Any]] = {"html": {}}

    def retrieve_supported_render_sources(cls) -> str:
        return "pipeline"

    def retrieve_supported_render_targets(self) -> Union[Set[str], str]:
        return "html"

    def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
        return [PipelineTransformer(kiara=self._kiara)]

    def retrieve_jinja_env(self) -> JinjaEnv:

        jinja_env = JinjaEnv(template_base="kiara")
        return jinja_env

    def get_template(self, render_config: RenderInputsSchema) -> Template:

        return self.get_jinja_env().get_template("pipeline/static_page/page.html.j2")

    def assemble_render_inputs(
        self, instance: Any, render_config: RenderInputsSchema
    ) -> Mapping[str, Any]:

        inputs = render_config.dict()
        inputs["pipeline"] = instance
        return inputs

Functions

retrieve_supported_render_sources() -> str
Source code in kiara/renderers/included_renderers/pipeline.py 
56
57
def retrieve_supported_render_sources(cls) -> str:
    return "pipeline"
retrieve_supported_render_targets() -> Union[Set[str], str]
Source code in kiara/renderers/included_renderers/pipeline.py 
59
60
def retrieve_supported_render_targets(self) -> Union[Set[str], str]:
    return "html"
retrieve_source_transformers() -> Iterable[SourceTransformer]
Source code in kiara/renderers/included_renderers/pipeline.py 
62
63
def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
    return [PipelineTransformer(kiara=self._kiara)]
retrieve_jinja_env() -> JinjaEnv
Source code in kiara/renderers/included_renderers/pipeline.py 
65
66
67
68
def retrieve_jinja_env(self) -> JinjaEnv:

    jinja_env = JinjaEnv(template_base="kiara")
    return jinja_env
get_template(render_config: RenderInputsSchema) -> Template
Source code in kiara/renderers/included_renderers/pipeline.py 
70
71
72
def get_template(self, render_config: RenderInputsSchema) -> Template:

    return self.get_jinja_env().get_template("pipeline/static_page/page.html.j2")
assemble_render_inputs(instance: Any, render_config: RenderInputsSchema) -> Mapping[str, Any]
Source code in kiara/renderers/included_renderers/pipeline.py 
74
75
76
77
78
79
80
def assemble_render_inputs(
    self, instance: Any, render_config: RenderInputsSchema
) -> Mapping[str, Any]:

    inputs = render_config.dict()
    inputs["pipeline"] = instance
    return inputs

PipelineRendererMarkdown

Bases: BaseJinjaRenderer[Type[Pipeline], RenderInputsSchema]

Renders a pipeline structure as static html page.

Source code in kiara/renderers/included_renderers/pipeline.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class PipelineRendererMarkdown(BaseJinjaRenderer[Type[Pipeline], RenderInputsSchema]):
    """Renders a pipeline structure as static html page."""

    _renderer_name = "pipeline_markdown"

    # _render_profiles: Mapping[str, Mapping[str, Any]] = {"html": {}}

    def retrieve_supported_render_sources(cls) -> str:
        return "pipeline"

    def retrieve_supported_render_targets(self) -> Union[Set[str], str]:
        return "markdown"

    def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
        return [PipelineTransformer(kiara=self._kiara)]

    def retrieve_jinja_env(self) -> JinjaEnv:

        jinja_env = JinjaEnv(template_base="kiara")
        return jinja_env

    def get_template(self, render_config: RenderInputsSchema) -> Template:

        return self.get_jinja_env().get_template("pipeline/markdown/pipeline.md.j2")

    def assemble_render_inputs(
        self, instance: Any, render_config: RenderInputsSchema
    ) -> Mapping[str, Any]:

        inputs = render_config.dict()
        inputs["pipeline"] = instance
        return inputs

Functions

retrieve_supported_render_sources() -> str
Source code in kiara/renderers/included_renderers/pipeline.py 
90
91
def retrieve_supported_render_sources(cls) -> str:
    return "pipeline"
retrieve_supported_render_targets() -> Union[Set[str], str]
Source code in kiara/renderers/included_renderers/pipeline.py 
93
94
def retrieve_supported_render_targets(self) -> Union[Set[str], str]:
    return "markdown"
retrieve_source_transformers() -> Iterable[SourceTransformer]
Source code in kiara/renderers/included_renderers/pipeline.py 
96
97
def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
    return [PipelineTransformer(kiara=self._kiara)]
retrieve_jinja_env() -> JinjaEnv
Source code in kiara/renderers/included_renderers/pipeline.py 
 99
100
101
102
def retrieve_jinja_env(self) -> JinjaEnv:

    jinja_env = JinjaEnv(template_base="kiara")
    return jinja_env
get_template(render_config: RenderInputsSchema) -> Template
Source code in kiara/renderers/included_renderers/pipeline.py 
104
105
106
def get_template(self, render_config: RenderInputsSchema) -> Template:

    return self.get_jinja_env().get_template("pipeline/markdown/pipeline.md.j2")
assemble_render_inputs(instance: Any, render_config: RenderInputsSchema) -> Mapping[str, Any]
Source code in kiara/renderers/included_renderers/pipeline.py 
108
109
110
111
112
113
114
def assemble_render_inputs(
    self, instance: Any, render_config: RenderInputsSchema
) -> Mapping[str, Any]:

    inputs = render_config.dict()
    inputs["pipeline"] = instance
    return inputs

PipelineRendererPngConfig

Bases: KiaraRendererConfig

Source code in kiara/renderers/included_renderers/pipeline.py 
117
118
119
120
121
class PipelineRendererPngConfig(KiaraRendererConfig):

    graph_type: Literal["execution", "data-flow", "data-flow-simple", "stages"] = Field(
        description="The type of graph to render."
    )

Attributes

graph_type: Literal['execution', 'data-flow', 'data-flow-simple', 'stages'] = Field(description='The type of graph to render.') class-attribute

PipelineInputsSchema

Bases: RenderInputsSchema

Source code in kiara/renderers/included_renderers/pipeline.py 
124
125
126
127
128
class PipelineInputsSchema(RenderInputsSchema):

    stages_extraction_type: str = Field(
        description="The type of stages extraction to use.", default="late"
    )

Attributes

stages_extraction_type: str = Field(description='The type of stages extraction to use.', default='late') class-attribute

PipelineRendererPng

Bases: KiaraRenderer[Pipeline, PipelineInputsSchema, bytes, PipelineRendererPngConfig]

Source code in kiara/renderers/included_renderers/pipeline.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
class PipelineRendererPng(
    KiaraRenderer[Pipeline, PipelineInputsSchema, bytes, PipelineRendererPngConfig]
):

    _renderer_name = "pipeline_png"
    _renderer_config_cls = PipelineRendererPngConfig  # type: ignore
    _inputs_schema = PipelineInputsSchema

    _renderer_profiles: Mapping[str, Mapping[str, Any]] = {
        "execution-graph-image": {"graph_type": "execution"},
        "data-flow-graph-image": {"graph_type": "data-flow"},
        "data-flow-simple-graph-image": {"graph_type": "data-flow-simple"},
        "stages-graph-image": {"graph_type": "stages"},
    }

    def retrieve_doc(self) -> Union[str, None]:

        graph_type = self.renderer_config.graph_type

        if graph_type == "execution":
            graph = "the execution graph"
        elif graph_type == "data-flow":
            graph = "the data flow graph"
        elif graph_type == "data-flow-simple":
            graph = "a simplified version of the data flow graph"
        elif graph_type == "stages":
            graph = "the stages graph"
        else:
            raise Exception(f"Invalid graph type: {graph_type}")

        return f"Render {graph} of the given pipeline as a image (in png format)."

    def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
        return [PipelineTransformer(kiara=self._kiara)]

    def get_renderer_alias(self) -> str:
        return f"{self.renderer_config.graph_type}_png"

    def retrieve_supported_render_sources(self) -> str:
        return "pipeline"

    def retrieve_supported_render_targets(self) -> Union[Iterable[str], str]:
        return f"{self.renderer_config.graph_type}_png"

    def _render(self, instance: Pipeline, render_config: PipelineInputsSchema) -> bytes:

        graph_type = self.renderer_config.graph_type

        if graph_type == "execution":
            graph = instance.structure.execution_graph
        elif graph_type == "data-flow":
            graph = instance.structure.data_flow_graph
        elif graph_type == "data-flow-simple":
            graph = instance.structure.data_flow_graph_simple
        elif graph_type == "stages":
            graph = instance.structure.get_stages_graph(
                stages_extraction_type=render_config.stages_extraction_type
            )
        else:
            raise Exception(f"Invalid graph type: {graph_type}")

        image = create_image(graph)
        return image

Functions

retrieve_doc() -> Union[str, None]
Source code in kiara/renderers/included_renderers/pipeline.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def retrieve_doc(self) -> Union[str, None]:

    graph_type = self.renderer_config.graph_type

    if graph_type == "execution":
        graph = "the execution graph"
    elif graph_type == "data-flow":
        graph = "the data flow graph"
    elif graph_type == "data-flow-simple":
        graph = "a simplified version of the data flow graph"
    elif graph_type == "stages":
        graph = "the stages graph"
    else:
        raise Exception(f"Invalid graph type: {graph_type}")

    return f"Render {graph} of the given pipeline as a image (in png format)."
retrieve_source_transformers() -> Iterable[SourceTransformer]
Source code in kiara/renderers/included_renderers/pipeline.py 
163
164
def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
    return [PipelineTransformer(kiara=self._kiara)]
get_renderer_alias() -> str
Source code in kiara/renderers/included_renderers/pipeline.py 
166
167
def get_renderer_alias(self) -> str:
    return f"{self.renderer_config.graph_type}_png"
retrieve_supported_render_sources() -> str
Source code in kiara/renderers/included_renderers/pipeline.py 
169
170
def retrieve_supported_render_sources(self) -> str:
    return "pipeline"
retrieve_supported_render_targets() -> Union[Iterable[str], str]
Source code in kiara/renderers/included_renderers/pipeline.py 
172
173
def retrieve_supported_render_targets(self) -> Union[Iterable[str], str]:
    return f"{self.renderer_config.graph_type}_png"

Functions