ads.pipeline.visualizer package¶

Submodules¶

ads.pipeline.visualizer.base module¶

class ads.pipeline.visualizer.base.GraphOrientation[source]¶

Bases: object

LEFT_RIGHT = 'LR'¶

TOP_BOTTOM = 'TB'¶

class ads.pipeline.visualizer.base.PipelineRenderer[source]¶

Bases: ABC

The base class responsible for the vizualizing a pipleine.

abstract render(steps: List[RendererItem], deps: Dict[str, List[RendererItem]], step_status: Dict[str, List[RendererItemStatus]] | None = None, **kwargs)[source]¶: Renders pipeline run.

save_to(*args, **kwargs) → str[source]¶: Saves the pipeline visualization to the provided format.

class ads.pipeline.visualizer.base.PipelineVisualizer(pipeline: Pipeline = None, pipeline_run: PipelineRun = None, renderer: PipelineRenderer = None)[source]¶

Bases: object

PipelineVisualizer class to visualize pipeline in text or graph.

pipeline¶

Pipeline instance.

Type:: Pipeline

pipeline_run¶

PipelineRun instance.

Type:: PipelineRun

steps¶

A list of RendererItem objects.

Type:: List[RendererItem]

deps¶

A dictionary mapping the key of a RendererItem to a list of RendererItem that this step depends on.

Type:: Dict[str, List[RendererItem]]

step_status¶

A dictionary mapping the key of a RendererItem to its current status.

Type:: Dict[str, RendererItemStatus], defaults to None.

Initialize a PipelineVisualizer object.

Parameters:

pipeline (Pipeline) – Pipeline instance.
pipeline_run (PipelineRun) – PipelineRun instance.
renderer (PipelineRenderer) – Renderer used to visualize pipeline in text or graph.

render(rankdir: str = 'TB')[source]¶

Renders pipeline step status.

Parameters:: rankdir (str, default to "TB".) – Direction of the rendered graph; allowed Values are {“TB”, “LR”}.
Return type:: None
Raises:: PipelineVisualizerError – If pipeline or renderer not specified.

to_svg(uri: str | None = None, rankdir: str = 'TB', **kwargs) → str[source]¶

Renders pipeline as graph in SVG string.

Parameters:

uri ((string, optional). Defaults to None.) – URI location to save the SVG string.
rankdir (str, default to "TB".) – Direction of the rendered graph; allowed Values are {“TB”, “LR”}.

Returns:

Graph in svg format.

Return type:

str

Raises:

PipelineVisualizerError – If pipeline or renderer not specified.

with_pipeline(value: Pipeline) → PipelineVisualizer[source]¶

Adds a Pipeline instance to be rendered.

Parameters:: value (Pipeline) – Pipeline instance.
Returns:: The PipelineVisualizer instance.
Return type:: PipelineVisualizer
Raises:: PipelineVisualizerError – If pipeline not specified.

with_pipeline_run(value: PipelineRun) → PipelineVisualizer[source]¶

Adds a PipelineRun instance to be rendered.

Parameters:: value (PipelineRun) – PipelineRun instance.
Returns:: The PipelineVisualizer instance.
Return type:: PipelineVisualizer
Raises:: PipelineVisualizerError – If pipeline run not specified.

with_renderer(value: PipelineRenderer) → PipelineVisualizer[source]¶

Add renderer to visualize pipeline.

Parameters:: value (object) – Renderer used to visualize pipeline in text or graph.
Returns:: The PipelineVisualizer instance.
Return type:: PipelineVisualizer
Raises:: PipelineVisualizerError – If renderer not specified.

exception ads.pipeline.visualizer.base.PipelineVisualizerError[source]¶: Bases: Exception

class ads.pipeline.visualizer.base.RendererItem(name: str, kind: str = '', spec: ForwardRef('Pipeline') | ForwardRef('PipelineStep') = None, _key: str = '')[source]¶

Bases: object

property key: str¶

Key of the item.

Returns:: The key of the item.
Return type:: str

kind: str = ''¶

name: str¶

spec: Pipeline | PipelineStep = None¶

class ads.pipeline.visualizer.base.RendererItemStatus(name: str, kind: str = '', time_started: datetime | None = None, time_finished: datetime | None = None, lifecycle_state: str = '', lifecycle_details: str = '', _key: str = '')[source]¶

Bases: object

Class represents the state of the renderer item.

property duration: int¶

Calculates duration in seconds between time_started and time_finished.

Returns:: The duration in seconds between time_started and time_finished.
Return type:: int

static format_datetime(value: datetime, format='%Y-%m-%d %H:%M:%S') → str[source]¶

Converts datetime object into a given format in string

Parameters:: dt (datetime.datetime) – Datetime object to be formated.
Returns:: A timestamp in a string format.
Return type:: str

classmethod from_pipeline_run(pipeline_run: PipelineRun) → RendererItemStatus[source]¶

Creates class instance from the PipelineRun object.

Parameters:: pipeline_run (PipelineRun) – The PipelineRun object.
Returns:: Instance of RendererItemStatus.
Return type:: RendererItemStatus

classmethod from_pipeline_step_run(pipeline_step_run: PipelineStepRun) → RendererItemStatus[source]¶

Creates class instance from the PipelineStepRun object.

Parameters:: pipeline_run (PipelineStepRun) – The PipelineStepRun object.
Returns:: Instance of RendererItemStatus.
Return type:: RendererItemStatus

property key: str¶

Key of the item.

Returns:: The key of the item.
Return type:: str

kind: str = ''¶

lifecycle_details: str = ''¶

lifecycle_state: str = ''¶

name: str¶

time_finished: datetime = None¶

time_started: datetime = None¶

class ads.pipeline.visualizer.base.StepKind[source]¶

Bases: object

CUSTOM_SCRIPT = 'CUSTOM_SCRIPT'¶

ML_JOB = 'ML_JOB'¶

PIPELINE = 'pipeline'¶

class ads.pipeline.visualizer.base.StepStatus[source]¶

Bases: object

ACCEPTED = 'ACCEPTED'¶

CANCELED = 'CANCELED'¶

CANCELING = 'CANCELING'¶

DELETED = 'DELETED'¶

FAILED = 'FAILED'¶

IN_PROGRESS = 'IN_PROGRESS'¶

SKIPPED = 'SKIPPED'¶

SUCCEEDED = 'SUCCEEDED'¶

WAITING = 'WAITING'¶

ads.pipeline.visualizer.graph_renderer module¶

class ads.pipeline.visualizer.graph_renderer.PipelineGraphRenderer(show_status: bool = False)[source]¶

Bases: PipelineRenderer

Initialize a PipelineGraphRenderer class.

Parameters:: show_status (bool, defaults to False.) – Whether to display status for steps.
Returns:: Nothing.
Return type:: None

render(steps: List[RendererItem], deps: Dict[str, List[RendererItem]] | None = None, step_status: Dict[str, RendererItemStatus] | None = None, rankdir: str = 'TB', **kwargs)[source]¶

Renders Pipeline graph.

Parameters:

steps (List[RendererItem]) – A list of RendererItem objects.
deps (Dict[str, List[RendererItem]]) – A dictionary mapping the key of a RendererItem to a list of RendererItem that this step depends on.
step_status (Dict[str, RendererItemStatus], defaults to None.) – A dictionary mapping the key of a RendererItem to its current status.
rankdir (str, default to "TB".) – Direction of the rendered graph; allowed Values are {“TB”, “LR”}.

Return type:

None

save_to(steps: List[RendererItem], deps: Dict[str, List[RendererItem]] | None = None, step_status: Dict[str, RendererItemStatus] | None = None, rankdir: str = 'TB', uri: str | None = None, format: str = 'svg', **kwargs) → str[source]¶

Renders pipeline as graph in selected format.

steps: List[RendererItem]: A list of RendererItem objects.
deps: Dict[str, List[RendererItem]]: A dictionary mapping the key of a RendererItem to a list of RendererItem that this step depends on.
step_status: Dict[str, RendererItemStatus], defaults to None.: A dictionary mapping the key of a RendererItem to its current status.
rankdir: str, default to “TB”.: Direction of the rendered graph; allowed Values are {“TB”, “LR”}.
uri: (string, optional). Defaults to None.: URI location to save the SVG string.
format: (str, optional). Defaults to “svg”.: The format to save the graph. Supported formats: “svg”, “html”.

Returns:: Graph in selected format.
Return type:: str

class ads.pipeline.visualizer.graph_renderer.RenderTo[source]¶

Bases: str

JPEG = 'jpeg'¶

PNG = 'png'¶

SVG = 'svg'¶

ads.pipeline.visualizer.text_renderer module¶

class ads.pipeline.visualizer.text_renderer.PipelineTextRenderer[source]¶

Bases: PipelineRenderer

render(steps: List[RendererItem], deps: Dict[str, List[RendererItem]], step_status: Dict[str, RendererItemStatus] | None = None, **kwargs)[source]¶

Render pipeline step status in text.

Parameters:

steps (List[RendererItem]) – A list of RendererItem objects.
deps (Dict[str, List[RendererItem]]) – A dictionary mapping the key of a RendererItem to a list of RendererItem that this step depends on.
step_status (Dict[str, RendererItemStatus], defaults to None.) – A dictionary mapping the key of a RendererItem to its current status.

Return type:

None

ads.pipeline.visualizer package¶

Submodules¶

ads.pipeline.visualizer.base module¶

ads.pipeline.visualizer.graph_renderer module¶

ads.pipeline.visualizer.text_renderer module¶

Module contents¶