layout_optimizer

layout_optimizer

DomoDataflow Canvas Layout Optimizer

Computes a layered (topological) layout for dataflow canvas tiles, applies consistent tile colours by action type, and optionally groups tiles into sections by upstream root.

Usage

from .layout_optimizer import optimize_canvas canvas = optimize_canvas(definition, add_sections=True)

auto_section

auto_section(
    canvas: CanvasElements,
    actions: Sequence,
    edges: list[tuple[str, str]],
) -> None

Group tiles into sections by their upstream root.

A root tile is one with no upstream dependencies. All downstream tiles that trace back to the same root are placed in a single :class:CanvasSection.

Sections are positioned to encompass their tiles with padding, and tiles inside sections use relative coordinates (relative to the section's top-left corner).

Parameters:

Name	Type	Description	Default
canvas	CanvasElements	The canvas element manager (mutated in-place).	required
actions	Sequence	Sequence of action objects (must have id attribute).	required
edges	list[tuple[str, str]]	List of (source_id, target_id) pairs.	required

Source code in src/crew_dcs/classes/DomoDataflow/layout_optimizer.py

def auto_section(  # noqa: C901
    canvas: CanvasElements,
    actions: Sequence,
    edges: list[tuple[str, str]],
) -> None:
    """Group tiles into sections by their upstream root.

    A *root* tile is one with no upstream dependencies.  All downstream
    tiles that trace back to the same root are placed in a single
    :class:`CanvasSection`.

    Sections are positioned to encompass their tiles with padding, and
    tiles inside sections use **relative** coordinates (relative to the
    section's top-left corner).

    Args:
        canvas: The canvas element manager (mutated in-place).
        actions: Sequence of action objects (must have ``id`` attribute).
        edges: List of ``(source_id, target_id)`` pairs.
    """
    tile_ids = {t.id for t in canvas.tiles}
    {a.id for a in actions}

    # Build reverse adjacency (child → parents)
    parents: dict[str, list[str]] = defaultdict(list)
    for src, tgt in edges:
        if src in tile_ids and tgt in tile_ids:
            parents[tgt].append(src)

    # Find roots — tiles with no upstream parents
    roots = [tid for tid in tile_ids if not parents.get(tid)]

    if not roots:
        # No roots means either no tiles or a pure cycle — nothing to section
        return

    # For each tile, trace back to find its root(s)
    # Use BFS upward; if multiple roots, assign to the first found
    tile_to_root: dict[str, str] = {}

    def _find_root(tid: str, visited: set[str] | None = None) -> str | None:
        visited = visited or set()
        if tid in visited:
            return None  # cycle
        visited.add(tid)
        if tid in set(roots):
            return tid
        for pid in parents.get(tid, []):
            r = _find_root(pid, visited)
            if r is not None:
                return r
        return None

    for tid in tile_ids:
        root = _find_root(tid)
        if root is not None:
            tile_to_root[tid] = root

    # Group tiles by root
    root_groups: dict[str, list[str]] = defaultdict(list)
    for tid, root in tile_to_root.items():
        root_groups[root].append(tid)

    # If only one group and it contains all tiles, skip sectioning
    # (a single section wrapping everything is not useful)
    if len(root_groups) <= 1:
        return

    # Remove existing sections before adding new ones
    canvas.remove_elements_where(lambda e: isinstance(e, CanvasSection))

    # Build tile position lookup (absolute positions)
    tile_pos: dict[str, tuple[int, int]] = {}
    for tile in canvas.tiles:
        abs_x, abs_y = canvas.abs_pos(tile)
        tile_pos[tile.id] = (abs_x, abs_y)

    # Sort roots for deterministic section ordering
    sorted_roots = sorted(root_groups.keys())

    # Compute section bounds and create sections
    # Sections are stacked vertically to avoid overlap
    section_y_offset = 0
    for idx, root in enumerate(sorted_roots):
        group = root_groups[root]
        if not group:
            continue

        # Compute bounding box of tiles in this group
        xs = [tile_pos[tid][0] for tid in group if tid in tile_pos]
        ys = [tile_pos[tid][1] for tid in group if tid in tile_pos]
        if not xs or not ys:
            continue

        min_x = min(xs)
        min_y = min(ys)
        max_x = max(xs)
        max_y = max(ys)

        # Section dimensions with padding (computed before offset)
        sec_x = min_x - _SECTION_PADDING
        sec_y_base = min_y - _SECTION_PADDING
        sec_w = (max_x + TILE_W + _SECTION_PADDING) - sec_x
        sec_h = (max_y + TILE_H + _SECTION_PADDING) - sec_y_base

        # Apply vertical offset for stacking
        sec_y = sec_y_base + section_y_offset

        # Pick a distinct colour
        bg_color = _SECTION_COLORS[idx % len(_SECTION_COLORS)]

        # Derive a section name from the root action id
        root_action = next((a for a in actions if a.id == root), None)
        sec_name = root_action.name if root_action and root_action.name else root

        section = canvas.add_section(
            name=sec_name,
            x=sec_x,
            y=sec_y,
            width=sec_w,
            height=sec_h,
            background_color=bg_color,
        )

        # Move tiles into the section (convert to relative coords)
        for tid in group:
            tile = canvas.get(tid)
            if tile and isinstance(tile, CanvasTile):
                # Convert absolute position to section-relative
                tile.x = tile_pos[tid][0] - sec_x
                tile.y = tile_pos[tid][1] - sec_y
                tile.parent_id = section.id

        # Advance y offset for next section
        section_y_offset += sec_h + _SECTION_GAP

optimize_canvas

optimize_canvas(
    definition, add_sections: bool = True
) -> CanvasElements

Optimize the canvas layout for a dataflow definition.

Computes a topological layout, applies standard tile colours, and optionally groups tiles into sections by upstream root.

Parameters:

Name	Type	Description	Default
definition		A :class:DomoDataflow_Definition instance. Must have Canvas, Actions, and canvas_edges() available.	required
add_sections	bool	Whether to create auto-sections (default True).	True

Returns:

Type	Description
CanvasElements	The modified :class:CanvasElements instance.

Example::

canvas = optimize_canvas(dataflow.Definition, add_sections=True)
await dataflow.Definition.save(version_note="Optimized canvas layout")

Source code in src/crew_dcs/classes/DomoDataflow/layout_optimizer.py

def optimize_canvas(
    definition,
    add_sections: bool = True,
) -> CanvasElements:
    """Optimize the canvas layout for a dataflow definition.

    Computes a topological layout, applies standard tile colours, and
    optionally groups tiles into sections by upstream root.

    Args:
        definition: A :class:`DomoDataflow_Definition` instance. Must have
            ``Canvas``, ``Actions``, and ``canvas_edges()`` available.
        add_sections: Whether to create auto-sections (default True).

    Returns:
        The modified :class:`CanvasElements` instance.

    Example::

        canvas = optimize_canvas(dataflow.Definition, add_sections=True)
        await dataflow.Definition.save(version_note="Optimized canvas layout")
    """
    actions = definition.Actions.actions

    # Compute edges from actions directly (not from canvas_edges())
    # because canvas_edges() requires tiles to already exist on the
    # canvas, which may not be the case when we're building from scratch.
    action_ids = {a.id for a in actions}
    edges = [
        (dep_id, action.id)
        for action in actions
        for dep_id in (action.depends_on or [])
        if dep_id in action_ids and action.id in action_ids
    ]

    # Ensure Canvas exists
    if definition.Canvas is None:
        definition.Canvas = CanvasElements()

    canvas = definition.Canvas

    # Compute positions
    positions = topological_layout(actions, edges)

    # Apply positions to existing tiles or create new ones
    existing_tile_ids = {t.id for t in canvas.tiles}

    # Remove tiles for actions that no longer exist
    canvas.keep_only_tiles(action_ids)

    # Apply positions — update existing tiles, create missing ones
    for action in actions:
        aid = action.id
        if aid not in positions:
            continue
        x, y = positions[aid]

        if aid in existing_tile_ids:
            # Update existing tile position
            tile = canvas.get(aid)
            if tile and isinstance(tile, CanvasTile):
                tile.x = x
                tile.y = y
                tile.parent_id = None  # reset section membership
        else:
            # Create new tile
            tile = CanvasTile.from_action(action_id=aid, x=x, y=y)
            canvas.add(tile)

    # Standardize colours
    standardize_tile_colors(canvas)

    # Auto-section
    if add_sections:
        auto_section(canvas, actions, edges)

    return canvas

standardize_tile_colors

standardize_tile_colors(canvas: CanvasElements) -> None

Apply consistent colours to all tiles based on action type.

Mutates the color and color_source attributes of each :class:CanvasTile in canvas.

Parameters:

Name	Type	Description	Default
canvas	CanvasElements	The canvas element manager to update in-place.	required

Source code in src/crew_dcs/classes/DomoDataflow/layout_optimizer.py

def standardize_tile_colors(canvas: CanvasElements) -> None:
    """Apply consistent colours to all tiles based on action type.

    Mutates the ``color`` and ``color_source`` attributes of each
    :class:`CanvasTile` in *canvas*.

    Args:
        canvas: The canvas element manager to update in-place.
    """
    for tile in canvas.tiles:
        color = _action_type_color(tile.id)
        tile.color = color
        tile.color_source = color

topological_layout

topological_layout(
    actions: Sequence, edges: list[tuple[str, str]]
) -> dict[str, tuple[int, int]]

Compute a layered layout from a DAG of actions and edges.

Uses Kahn's algorithm for topological sort to assign layers (depth). Tiles that share the same layer are stacked vertically.

Parameters:

Name	Type	Description	Default
actions	Sequence	Sequence of action objects (must have id attribute).	required
edges	list[tuple[str, str]]	List of (source_id, target_id) pairs.	required

Returns:

Type	Description
dict[str, tuple[int, int]]	Dict mapping action_id → (x, y) pixel coordinates.

Notes

• Cycles are broken by skipping nodes that remain after the topological sort completes (they are placed in the last layer).
• Nodes with no upstream edges are placed in layer 0.

Source code in src/crew_dcs/classes/DomoDataflow/layout_optimizer.py

def topological_layout(  # noqa: C901
    actions: Sequence,
    edges: list[tuple[str, str]],
) -> dict[str, tuple[int, int]]:
    """Compute a layered layout from a DAG of actions and edges.

    Uses Kahn's algorithm for topological sort to assign layers (depth).
    Tiles that share the same layer are stacked vertically.

    Args:
        actions: Sequence of action objects (must have ``id`` attribute).
        edges: List of ``(source_id, target_id)`` pairs.

    Returns:
        Dict mapping ``action_id`` → ``(x, y)`` pixel coordinates.

    Notes:
        - Cycles are broken by skipping nodes that remain after the
          topological sort completes (they are placed in the last layer).
        - Nodes with no upstream edges are placed in layer 0.
    """
    action_ids = [a.id for a in actions]
    id_set = set(action_ids)

    # Build adjacency and in-degree
    children: dict[str, list[str]] = defaultdict(list)
    in_degree: dict[str, int] = defaultdict(int)
    for src, tgt in edges:
        if src in id_set and tgt in id_set:
            children[src].append(tgt)
            in_degree[tgt] += 1

    # Ensure every action has an entry in in_degree
    for aid in action_ids:
        in_degree.setdefault(aid, 0)

    # Kahn's algorithm — assign layers
    layers: dict[str, int] = {}
    queue: deque[str] = deque(aid for aid in action_ids if in_degree[aid] == 0)
    visited: set[str] = set()

    while queue:
        node = queue.popleft()
        if node in visited:
            continue
        visited.add(node)

        # Layer = max(parent layers) + 1, or 0 for roots
        parent_layers = [
            layers[pid]
            for pid in id_set
            if pid != node and node in children.get(pid, [])
        ]
        layers[node] = max(parent_layers, default=-1) + 1

        for child in children[node]:
            in_degree[child] -= 1
            if in_degree[child] == 0:
                queue.append(child)

    # Handle cycle nodes — place them in the deepest layer + 1
    cycle_nodes = id_set - visited
    if cycle_nodes:
        max_layer = max(layers.values(), default=0) + 1
        for node in cycle_nodes:
            layers[node] = max_layer

    # Group by layer and assign positions
    layer_groups: dict[int, list[str]] = defaultdict(list)
    for aid, layer in layers.items():
        layer_groups[layer].append(aid)

    # Sort within each layer for deterministic output
    for layer in layer_groups:
        layer_groups[layer].sort()

    positions: dict[str, tuple[int, int]] = {}
    for layer_idx in sorted(layer_groups):
        group = layer_groups[layer_idx]
        for pos_idx, aid in enumerate(group):
            x = layer_idx * (TILE_W + _LAYER_GAP)
            y = pos_idx * (TILE_H + _ROW_GAP)
            positions[aid] = (x, y)

    return positions