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: Union[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

Module contents#