Layout

Canvas layout and component positioning

NiFi Canvas Layout Module.

Provides a block-based system for positioning components on the NiFi canvas. Component dimensions are empirically derived from the NiFi UI, and all positions snap to NiFi’s 8-pixel grid for UI compatibility.

Two Layout Patterns:

  1. FLOW LAYOUT - For components with connections (processors, funnels, ports) Components need padding for queue labels and retry loops. Use: - below(), above(), left_of(), right_of() for sequential flow building - fork() for diagonal side branches - new_flow() for starting independent flows with separation

  2. PROCESS GROUP GRID - For organizing PGs without connections PGs are packed in a simple grid with minimal padding. Use: - align_pg_grid() to arrange existing PGs into a grid - suggest_pg_position() to find the next slot in a grid

Block System:

  • BLOCK_WIDTH (400px): Horizontal spacing for grid layouts

  • BLOCK_HEIGHT (200px): Vertical spacing (includes room for connection queues)

  • FORK_SPACING (640px): Diagonal fork spacing (processor + queue box + padding)

Component Dimensions (empirically verified):

  • Processor: 352 x 128 px

  • Process Group: 384 x 176 px

  • Port (Input/Output): 240 x 48 px

  • Funnel: 48 x 48 px

  • Queue Box: 224 x 56 px (connection label)

  • Label: User-definable (default 150 x 150 px)

Flow Layout Example:

import nipyapi

# Start a new flow on empty canvas (returns DEFAULT_ORIGIN: 400, 400)
first_pos = nipyapi.layout.new_flow()
proc1 = nipyapi.canvas.create_processor(pg, proc_type, location=first_pos)

# Create next processor below
proc2 = nipyapi.canvas.create_processor(pg, proc_type,
    location=nipyapi.layout.below(proc1))

# Place a funnel centered below (for smaller components)
funnel = nipyapi.canvas.create_funnel(pg.id,
    position=nipyapi.layout.below(proc2, align="center"))

# Fork a side branch (diagonal: right and down)
side_proc = nipyapi.canvas.create_processor(pg, proc_type,
    location=nipyapi.layout.fork(proc1, direction="right"))

Process Group Grid Example:

# Arrange all PGs in root canvas into a sorted grid
nipyapi.layout.align_pg_grid(root_id, sort_by_name=True)

# Add a new PG in the next available grid slot
pos = nipyapi.layout.suggest_pg_position(root_id)
nipyapi.canvas.create_process_group(root_pg, "New PG", location=pos)

Automatic Flow Layout:

plan = nipyapi.layout.suggest_flow_layout(pg.id)
for item in plan['spine'] + plan['branches']:
    comp = get_component(item['id'])
    nipyapi.layout.move_component(comp, item['position'])

This typically achieves 90% organization. See limitations below.

Limitations of suggest_flow_layout():

The automatic layout handles most cases well but has known limitations that require post-layout visual inspection and manual adjustment:

  1. TERMINAL BRANCH OVERLAPS: Multiple branches ending at similar depths may place components at overlapping positions. Fix by nudging components horizontally using move_component().

  2. FEEDBACK LOOP ROUTING: Long connections that loop back to earlier components (e.g., retry loops) create diagonal lines across the layout. The algorithm clears all bends for clean straight lines, but feedback loops may benefit from manual bend placement to route around components.

  3. QUEUE BOX COLLISIONS: Connection label boxes are not collision-detected. Some queue labels may overlap components or each other. Adjust by adding bends to reroute connections or moving components slightly.

  4. SEMANTIC VS LONGEST PATH: The spine is the longest forward path, which may include error handling rather than the “success” path. Use prefer_success=True in find_flow_spine() to weight success relationships, though significantly longer paths may still win.

Recommended workflow for complex flows:

  1. Run suggest_flow_layout() for initial organization

  2. Visual inspection (screenshot or UI review)

  3. Adjust overlapping terminals with move_component()

  4. Add bends to feedback loops for clean routing

  5. Verify queue label positions

nipyapi.layout.above(component, blocks: int = 1, align: str = 'aligned') → Tuple[float, float][source]

Calculate position above a component.

Parameters:

  • component – Reference component to position relative to.

  • blocks (int) – Number of blocks above (default 1).

  • align (str) – Horizontal alignment - “aligned” (same X, default), “center” (for funnels), or “center_port” (for ports).

Returns:

Position tuple (x, y) for the new component

nipyapi.layout.align_pg_grid(parent_pg_id: str, columns: int = None, sort_by_name: bool = False, origin: tuple = (400, 400), dry_run: bool = False)[source]

Align all process groups within a parent to a standard grid.

Parameters:

  • parent_pg_id – ID of the parent process group containing PGs to align

  • columns – Number of columns in the grid. If None (default), auto-calculates the optimal number to create a square-ish layout using ceil(sqrt(n))

  • sort_by_name – If True, sort PGs alphabetically by name (default False)

  • origin – Grid starting position (default (0, 0))

  • dry_run – If True, return planned moves without executing (default False)

Returns:

[{‘name’, ‘id’, ‘from’, ‘to’}, …]

Return type:

List of dicts with move details

Example:

# Auto-calculate columns, sort alphabetically
nipyapi.layout.align_pg_grid(pg_id, sort_by_name=True)

# Preview moves without executing
moves = nipyapi.layout.align_pg_grid(pg_id, dry_run=True)
for m in moves:
    print(f"{m['name']}: {m['from']} -> {m['to']}")
nipyapi.layout.below(component, blocks: int = 1, align: str = 'aligned') → Tuple[float, float][source]

Calculate position below a component.

Use this for building vertical flows where components connect top-to-bottom.

Parameters:

  • component – Reference component to position relative to.

  • blocks (int) – Number of blocks below (default 1).

  • align (str) – Horizontal alignment - “aligned” (same X, default), “center” (for funnels), or “center_port” (for ports).

Returns:

Position tuple (x, y) for the new component

Example:

proc2_pos = nipyapi.layout.below(proc1)
funnel_pos = nipyapi.layout.below(proc1, align="center")
port_pos = nipyapi.layout.below(proc1, align="center_port")
nipyapi.layout.clear_flow_bends(pg_id: str, include_self_loops: bool = False) → int[source]

Clear all bends from connections in a process group.

Use this before reorganizing a flow layout. Old bends look wrong after components are moved to new positions, so clearing them first ensures clean straight-line connections after the layout is applied.

By default, self-loop bends (retry loops) are preserved because they are required for the connection to render correctly in the NiFi UI.

Parameters:

  • pg_id – Process group ID containing the connections to clear

  • include_self_loops – If True, also clear bends on self-loop connections. Default False preserves self-loop shapes which are required for correct UI rendering.

Returns:

Number of connections that had bends cleared

Return type:

int

Example:

# Before reorganizing a messy flow
nipyapi.layout.clear_flow_bends(pg.id)

# Apply new layout
plan = nipyapi.layout.suggest_flow_layout(pg.id)
for item in plan['spine'] + plan['branches']:
    comp = get_component(item['id'])
    nipyapi.layout.move_component(comp, item['position'])
nipyapi.layout.find_flow_spine(pg_id: str, start_component=None, prefer_success: bool = False) → list[source]

Find the main artery (spine) of a flow - the longest chain of connections.

Flows typically branch like arteries: a main vertical path with smaller side branches. This function identifies the spine by finding the longest path from any entry point (component with no incoming connections) to any exit point (component with no outgoing connections).

The spine can be used to: - Lay out the main flow vertically - Identify side branches (components not on the spine) - Restructure a messy flow into a logical layout

Algorithm:

  1. Build directed graph from connections (skip self-loops)

  2. Track which edges use “success” relationship

  3. Find entry points (in-degree = 0)

  4. Use DFS to find best path based on strategy

  5. Return the path as a list of component IDs

Parameters:

  • pg_id – Process group ID containing the flow to analyze

  • start_component – Optional component to start from. If provided, finds the longest path starting from this component. If None, finds the globally longest path.

  • prefer_success – Strategy for choosing between paths. False (default) picks longest path with success count as tiebreaker, giving the simplest shape. True heavily weights “success” relationships, so shorter paths with more success edges may win.

Returns:

List of component IDs representing the spine, ordered from entry to exit. Returns empty list if no connections exist.

Example:

# Find the longest spine (default - simplest shape)
spine_ids = nipyapi.layout.find_flow_spine(pg.id)

# Find the semantic "success" spine
spine_ids = nipyapi.layout.find_flow_spine(pg.id, prefer_success=True)

# Get actual component entities
flow = nipyapi.canvas.get_flow(pg.id)
components_map = {p.id: p for p in flow.process_group_flow.flow.processors}
spine = [components_map[cid] for cid in spine_ids if cid in components_map]

# Lay out spine vertically
for i, comp in enumerate(spine):
    pos = nipyapi.layout.grid_position(row=i, col=0)
    nipyapi.layout.move_component(comp, pos)
nipyapi.layout.fork(component, direction: str = 'right', rows: int = 1) → Tuple[float, float][source]

Calculate position for a forked side branch from a flow.

Uses component-based spacing: processor width + queue box width + padding. This ensures the queue label on the diagonal connection doesn’t overlap either the source or the forked processor.

Parameters:

  • component – Source component (fork point)

  • direction – “right” or “left” (default “right”)

  • rows – Number of rows below source (default 1)

Returns:

Position tuple (x, y) for the forked processor

Example:

# Fork a side path from a processor
side_pos = nipyapi.layout.fork(main_proc, direction="right")
side_proc = nipyapi.canvas.create_processor(pg, proc_type, location=side_pos)
nipyapi.layout.get_canvas_bounds(pg_id: str = None, components: list = None) → dict[source]

Get the bounding box of components.

Can operate in two modes:

  1. pg_id provided: Get bounds of all components in the process group

  2. components provided: Get bounds of a specific list of components

Parameters:

  • pg_id – Process group ID (fetches all components in PG)

  • components – List of component entities to calculate bounds for

Returns:

min_x, max_x, min_y, max_y, width, height Returns None values if no components found.

Return type:

Dict with keys

Example:

# Get bounds of entire canvas
bounds = nipyapi.layout.get_canvas_bounds(pg_id=pg.id)

# Get bounds of a specific flow (from get_flow_components)
flow_components = nipyapi.canvas.get_flow_components(proc1)
bounds = nipyapi.layout.get_canvas_bounds(components=flow_components)
nipyapi.layout.get_pg_grid_position(row: int, col: int, origin: tuple = (400, 400)) → tuple[source]

Calculate position for a process group in a standard grid.

Uses BLOCK_WIDTH for horizontal spacing and BLOCK_HEIGHT for vertical spacing. This is typically used internally by align_pg_grid() and suggest_pg_position().

Parameters:

  • row – Row index (0-based, increases downward)

  • col – Column index (0-based, increases rightward)

  • origin – Grid origin position, top-left corner (default DEFAULT_ORIGIN)

Returns:

Position tuple (x, y) for the grid cell

Example:

# Get position for row 2, column 3
pos = get_pg_grid_position(2, 3)  # Returns (1600, 800) with default origin
nipyapi.layout.get_position(component) → Tuple[float, float][source]

Extract the (x, y) position from any canvas component.

Parameters:

component – Any NiFi canvas component (processor, process group, funnel, port, etc.)

Returns:

Tuple of (x, y) coordinates

Raises:

ValueError – If component has no position attribute

nipyapi.layout.get_side_branches(pg_id: str, spine: list = None, recursive: bool = True) → dict[source]

Find all side branches that fork off from the spine.

A side branch is any path that diverges from the main spine. This function recursively finds branches of branches, building a complete picture of the flow structure.

Parameters:

  • pg_id – Process group ID containing the flow

  • spine – List of component IDs representing the spine. If None, will be calculated using find_flow_spine().

  • recursive – If True (default), also find branches of branches. If False, only find direct branches from spine components.

Returns:

Dict mapping component IDs to lists of their branch component IDs. Keys are component IDs that have branches, values are lists of branch component IDs. If recursive=True, branch components that have their own sub-branches will also appear as keys.

Example:

spine = nipyapi.layout.find_flow_spine(pg.id)
branches = nipyapi.layout.get_side_branches(pg.id, spine)

# Layout: spine vertical, branches to the right
for i, comp_id in enumerate(spine):
    main_pos = nipyapi.layout.grid_position(row=i, col=0)
    # ... move main component

    for j, branch_id in enumerate(branches.get(comp_id, [])):
        branch_pos = nipyapi.layout.fork(main_component, direction="right")
        # ... move branch component
nipyapi.layout.grid_position(row: int, col: int, origin: tuple = (400, 400)) → tuple[source]

Calculate position in a grid layout.

Useful for creating organized layouts of related components.

Parameters:

  • row – Row index (0-based, increases downward)

  • col – Column index (0-based, increases rightward)

  • origin – Starting position for the grid (default (0, 0))

Returns:

Position tuple (x, y) for the grid cell

Example:

# Create a 2x2 grid of processors
for row in range(2):
    for col in range(2):
        pos = nipyapi.layout.grid_position(row, col)
        create_processor(pg, proc_type, location=pos)
nipyapi.layout.left_of(component, blocks: int = 1, align: str = 'aligned') → Tuple[float, float][source]

Calculate position to the left of a component.

Parameters:

  • component – Reference component to position relative to.

  • blocks (int) – Number of blocks to the left (default 1).

  • align (str) – Vertical alignment - “aligned” (same Y, default), “center” (for funnels), or “center_port” (for ports).

Returns:

Position tuple (x, y) for the new component

nipyapi.layout.move_component(component, position: tuple, refresh: bool = True, include_retry: bool = True)[source]

Move any canvas component to a new position.

Automatically detects the component type and calls the appropriate move function.

Parameters:

  • component – Any canvas component (processor, process group, funnel, port, label)

  • position – New (x, y) position tuple

  • refresh – Whether to refresh the component before updating (default True)

  • include_retry – If True (default), also move bends on retry (self-loop) connections to preserve their visual shape. This matches NiFi UI behavior when dragging a single component. Only applies to processors.

Returns:

Updated component entity

Example:

# Move a processor down by one block
new_pos = nipyapi.layout.below(proc1)
nipyapi.layout.move_component(proc1, new_pos)

# Move without adjusting retry bends (for batch operations)
nipyapi.layout.move_component(proc1, new_pos, include_retry=False)
nipyapi.layout.move_funnel(funnel, position: tuple, refresh: bool = True)[source]

Move a funnel to a new position.

Parameters:

  • funnel – FunnelEntity to move

  • position – New (x, y) position tuple

  • refresh – Whether to refresh the funnel before updating (default True)

Returns:

Updated FunnelEntity

nipyapi.layout.move_label(label, position: tuple, refresh: bool = True)[source]

Move a label to a new position.

Parameters:

  • label – LabelEntity to move

  • position – New (x, y) position tuple

  • refresh – Whether to refresh the label before updating (default True)

Returns:

Updated LabelEntity

nipyapi.layout.move_port(port, position: tuple, refresh: bool = True)[source]

Move an input or output port to a new position.

Parameters:

  • port – PortEntity to move (input or output)

  • position – New (x, y) position tuple

  • refresh – Whether to refresh the port before updating (default True)

Returns:

Updated PortEntity

nipyapi.layout.move_process_group(process_group, position: tuple, refresh: bool = True)[source]

Move a process group to a new position.

Parameters:

  • process_group – ProcessGroupEntity to move

  • position – New (x, y) position tuple

  • refresh – Whether to refresh the process group before updating (default True)

Returns:

Updated ProcessGroupEntity

nipyapi.layout.move_processor(processor, position: tuple, refresh: bool = True, include_retry: bool = True)[source]

Move a processor to a new position.

Parameters:

  • processor – ProcessorEntity to move

  • position – New (x, y) position tuple

  • refresh – Whether to refresh the processor before updating (default True)

  • include_retry – If True (default), also move bends on retry (self-loop) connections to preserve their visual shape. This matches NiFi UI behavior when dragging a processor.

Returns:

Updated ProcessorEntity

nipyapi.layout.new_flow(component=None, direction: str = 'right') → Tuple[float, float][source]

Calculate position for a new flow relative to a known component.

Use this when you have a specific reference component and want to start a new independent flow next to it. Creates 2 blocks (800px) separation to allow room for side-connections and retry loops on both flows.

If no component is provided (empty canvas), returns DEFAULT_ORIGIN.

When to use new_flow vs suggest_empty_position:

  • new_flow(): You know which component to start next to

  • suggest_empty_position(): Find empty space by scanning all components

Parameters:

  • component – Reference component (typically the top of an existing flow). If None, returns DEFAULT_ORIGIN for starting on an empty canvas.

  • direction – “right” or “left” (default “right”)

Returns:

Position tuple (x, y) for the new flow’s first component

Example:

# Start first flow on empty canvas
first_pos = nipyapi.layout.new_flow()  # Returns DEFAULT_ORIGIN

# Start a new flow to the right of an existing one
new_flow_pos = nipyapi.layout.new_flow(existing_flow_top)
proc1 = nipyapi.canvas.create_processor(pg, proc_type, location=new_flow_pos)
nipyapi.layout.right_of(component, blocks: int = 1, align: str = 'aligned') → Tuple[float, float][source]

Calculate position to the right of a component.

Use this for placing related components side-by-side within the same flow.

Parameters:

  • component – Reference component to position relative to.

  • blocks (int) – Number of blocks to the right (default 1).

  • align (str) – Vertical alignment - “aligned” (same Y, default), “center” (for funnels), or “center_port” (for ports).

Returns:

Position tuple (x, y) for the new component

Example:

proc2_pos = nipyapi.layout.right_of(proc1)
funnel_pos = nipyapi.layout.right_of(proc1, align="center")
nipyapi.layout.snap_position(position: Tuple[float, float]) → Tuple[int, int][source]

Snap a position tuple to the grid.

Parameters:

position – (x, y) tuple

Returns:

Grid-aligned (x, y) tuple

Example:

pos = snap_position((145, 203))  # Returns (144, 200)
nipyapi.layout.snap_to_grid(value: float) → int[source]

Snap a coordinate value to the nearest grid point.

NiFi’s UI uses an 8-pixel grid for drag-and-drop positioning. While the API accepts any coordinate value, aligning to the UI grid ensures users can reposition components manually without unexpected shifts.

Parameters:

value – Coordinate value to snap

Returns:

Value rounded to nearest multiple of GRID_SIZE (8)

Example:

x = snap_to_grid(145)  # Returns 144
y = snap_to_grid(150)  # Returns 152
nipyapi.layout.suggest_bend_position(source, target) → Tuple[float, float][source]

Calculate an optimal bend position for a connection between two components.

Creates a clean right-angle path by placing the bend at the intersection of: - Horizontal line from the source component’s exit point - Vertical line to the target component’s entry point

This produces an L-shaped connection that is geometrically clean and easy to follow visually.

Parameters:

  • source – Source component entity

  • target – Target component entity

Returns:

Position tuple (x, y) for the bend point

Example:

bend = nipyapi.layout.suggest_bend_position(proc1, proc2)
nipyapi.canvas.create_connection(proc1, proc2, bends=[bend])
nipyapi.layout.suggest_empty_position(pg_id: str, prefer: str = 'right') → tuple[source]

Find empty space by scanning all components in a process group.

Analyzes all existing components to find the bounding box, then suggests a position at the edge of existing content that won’t overlap.

When to use suggest_empty_position vs new_flow:

  • suggest_empty_position(): Find empty space without knowing a reference

  • new_flow(): Position relative to a specific known component

Parameters:

  • pg_id – Process group ID to scan for existing components

  • prefer – Direction to expand from existing content: - “right”: Next column to the right, same top row (default) - “left”: Next column to the left, same top row - “below”: Next row below, same left column - “above”: Next row above, same left column

Returns:

Position tuple (x, y) for new component. Returns DEFAULT_ORIGIN if the canvas is empty.

Example:

# Find empty space to the right of all existing flows
pos = nipyapi.layout.suggest_empty_position(pg.id, prefer="right")
proc = nipyapi.canvas.create_processor(pg, proc_type, location=pos)
nipyapi.layout.suggest_flow_layout(pg_id: str) → dict[source]

Analyze a flow and suggest an organized layout.

Returns a structured plan for laying out the flow with spine components vertically (column 0) and side branches to the right. The algorithm finds the spine, recursively finds all branches, then assigns grid positions.

Parameters:

pg_id – Process group ID containing the flow

Returns:

Dict with ‘spine’ and ‘branches’ keys. Each contains a list of dicts with ‘id’, ‘row’, ‘col’, and ‘position’ keys. Branch items also have ‘fork_from’ indicating the parent component ID.

Example:

plan = nipyapi.layout.suggest_flow_layout(pg.id)

for item in plan['spine'] + plan['branches']:
    comp = get_component_by_id(item['id'])
    nipyapi.layout.move_component(comp, item['position'])
nipyapi.layout.suggest_pg_position(parent_pg_id: str) → tuple[source]

Suggest a position for a new process group that doesn’t overlap existing ones.

Finds a position within a square-ish grid layout. Prefers filling gaps in existing rows before extending the grid.

For best results, first call align_pg_grid() to organize existing PGs, then this function will find gaps in that organized grid.

Parameters:

parent_pg_id – ID of the parent process group

Returns:

Position tuple (x, y) for the new process group

Example:

pos = nipyapi.layout.suggest_pg_position(root_pg_id)
new_pg = nipyapi.canvas.create_process_group(root, "New PG", location=pos)
nipyapi.layout.transpose_flow(components: list, offset: tuple, pg_id: str = None, connections=None)[source]

Move an entire flow by the given offset, including all connection bends.

This function handles the complexity of moving a flow as a unit. It moves all components by the offset, then moves bends on all connections within the flow (both retry loops and cross-component connections).

This matches the behavior of selecting multiple components in the NiFi UI and dragging them together.

Parameters:

  • components – List of component entities to move (from get_flow_components)

  • offset – Tuple (dx, dy) representing the movement offset

  • pg_id – Process group ID containing the flow. If None, inferred from first component.

  • connections – Optional list of ConnectionEntity objects. If provided, these connections will be used for bend updates (avoiding an API call). Typically obtained from get_flow_components().connections.

Returns:

List of updated component entities

Example:

# Get the complete flow subgraph (single API call)
flow = nipyapi.canvas.get_flow_components(start_proc)

# Move entire flow with connections pre-fetched (no additional API calls)
nipyapi.layout.transpose_flow(
    flow.components, offset=(400, 0), connections=flow.connections
)

# Or without pre-fetched connections (connections will be fetched)
nipyapi.layout.transpose_flow(flow.components, offset=(400, 0))