Dynamic Python Blocks - Roboflow Inference (original) (raw)

When the syntax for Workflow definitions was outlined, one key aspect was not covered: the ability to define blocks directly within the Workflow definition itself. This section can include the manifest and Python code for blocks defined in-place, which are dynamically interpreted by the Execution Engine. These in-place blocks function similarly to those statically defined in plugins, yet provide much more flexibility.

Warning

Dynamic blocks only work in your local deployment of inference and are not supported on the Roboflow hosted platform.

If you wish to disable the functionality, export ALLOW_CUSTOM_PYTHON_EXECUTION_IN_WORKFLOWS=False

Theory¶

The high-level overview of Dynamic Python blocks functionality:

• user provides definition of dynamic block in JSON
• definition contains information required by Execution Engine to construct WorkflowBlockManifest and WorkflowBlock out of the document
• in runtime, Compiler turns definition into dynamically created Python classes - exactly the same as statically defined blocks
• In Workflow definition, you may declare steps that use dynamic blocks, as if dynamic blocks were standard static ones

Example¶

Let's take a look and discuss example workflow with dynamic Python blocks.

Workflow with dynamic block

{ "version": "1.0", "inputs": [ { "type": "WorkflowImage", "name": "image" } ], "dynamic_blocks_definitions": [ { "type": "DynamicBlockDefinition", "manifest": { "type": "ManifestDescription", "block_type": "OverlapMeasurement", "inputs": { "predictions": { "type": "DynamicInputDefinition", "selector_types": [ "step_output" ] }, "class_x": { "type": "DynamicInputDefinition", "value_types": [ "string" ] }, "class_y": { "type": "DynamicInputDefinition", "value_types": [ "string" ] } }, "outputs": { "overlap": { "type": "DynamicOutputDefinition", "kind": [] } } }, "code": { "type": "PythonCode", "run_function_code": "\ndef run(self, predictions: sv.Detections, class_x: str, class_y: str) -> BlockResult:\n bboxes_class_x = predictions[predictions.data[\"class_name\"] == class_x]\n bboxes_class_y = predictions[predictions.data[\"class_name\"] == class_y]\n overlap = []\n for bbox_x in bboxes_class_x:\n bbox_x_coords = bbox_x[0]\n bbox_overlaps = []\n for bbox_y in bboxes_class_y:\n if bbox_y[-1][\"detection_id\"] == bbox_x[-1][\"detection_id\"]:\n continue\n bbox_y_coords = bbox_y[0]\n x_min = max(bbox_x_coords[0], bbox_y_coords[0])\n y_min = max(bbox_x_coords[1], bbox_y_coords[1])\n x_max = min(bbox_x_coords[2], bbox_y_coords[2])\n y_max = min(bbox_x_coords[3], bbox_y_coords[3])\n # compute the area of intersection rectangle\n intersection_area = max(0, x_max - x_min + 1) * max(0, y_max - y_min + 1)\n box_x_area = (bbox_x_coords[2] - bbox_x_coords[0] + 1) * (bbox_x_coords[3] - bbox_x_coords[1] + 1)\n local_overlap = intersection_area / (box_x_area + 1e-5)\n bbox_overlaps.append(local_overlap)\n overlap.append(bbox_overlaps)\n return {\"overlap\": overlap}\n" } }, { "type": "DynamicBlockDefinition", "manifest": { "type": "ManifestDescription", "block_type": "MaximumOverlap", "inputs": { "overlaps": { "type": "DynamicInputDefinition", "selector_types": [ "step_output" ] } }, "outputs": { "max_value": { "type": "DynamicOutputDefinition", "kind": [] } } }, "code": { "type": "PythonCode", "run_function_code": "\ndef run(self, overlaps: List[List[float]]) -> BlockResult:\n max_value = -1\n for overlap in overlaps:\n for overlap_value in overlap:\n if not max_value:\n max_value = overlap_value\n else:\n max_value = max(max_value, overlap_value)\n return {\"max_value\": max_value}\n" } } ], "steps": [ { "type": "RoboflowObjectDetectionModel", "name": "model", "image": "$inputs.image", "model_id": "yolov8n-640" }, { "type": "OverlapMeasurement", "name": "overlap_measurement", "predictions": "$steps.model.predictions", "class_x": "dog", "class_y": "dog" }, { "type": "ContinueIf", "name": "continue_if", "condition_statement": { "type": "StatementGroup", "statements": [ { "type": "BinaryStatement", "left_operand": { "type": "DynamicOperand", "operand_name": "overlaps", "operations": [ { "type": "SequenceLength" } ] }, "comparator": { "type": "(Number) >=" }, "right_operand": { "type": "StaticOperand", "value": 1 } } ] }, "evaluation_parameters": { "overlaps": "$steps.overlap_measurement.overlap" }, "next_steps": [ "$steps.maximum_overlap" ] }, { "type": "MaximumOverlap", "name": "maximum_overlap", "overlaps": "$steps.overlap_measurement.overlap" } ], "outputs": [ { "type": "JsonField", "name": "overlaps", "selector": "$steps.overlap_measurement.overlap" }, { "type": "JsonField", "name": "max_overlap", "selector": "$steps.maximum_overlap.max_value" } ] }

Let's start the analysis from dynamic_blocks_definitions - this is the part of Workflow Definition that provides a list of dynamic blocks. Each block contains two sections:

• manifest - providing JSON representation of BlockManifest - refer blocks development guide
• code - shipping Python code

Definition of block manifest¶

Manifest definition contains several fields, including:

• block_type - equivalent of type field in block manifest - must provide unique block identifier
• inputs - dictionary with names and definitions of dynamic inputs
• outputs - dictionary with names and definitions of dynamic outputs
• output_dimensionality_offset - field specifies output dimensionality
• accepts_batch_input - field dictates if input data in runtime is to be provided in batches by Execution Engine
• accepts_empty_values - field deciding if empty inputs will be ignored while constructing step inputs

In any doubt, refer to blocks development guide, as the dynamic blocks replicates standard blocs capabilities.

Definition of dynamic input¶

Dynamic inputs define fields of dynamically created block manifest. In other words, this is definition based on which BlockManifest class will be created in runtime.

Each input may define the following properties:

• has_default_value - flag to decide if dynamic manifest field has default
• default_value - default value (used only if has_default_value=True
• is_optional - flag to decide if dynamic manifest field is optional
• is_dimensionality_reference - flag to decide if dynamic manifest field ship selector to be used in runtime as dimensionality reference
• dimensionality_offset - dimensionality offset for configured input property of dynamic manifest
• selector_types - type of selectors that may be used by property (one of input_image, step_output_image, input_parameter, step_output). Step may not hold selector, but then must provide definition of specific type.
• selector_data_kind - dictionary with list of selector kinds specific for each selector type
• value_types - definition of specific type that is to be placed in manifest - this field specifies typing of dynamically created manifest fields w.r.t Python types. Selection of types: any, integer, float, boolean, dict, list, strig

Definition of dynamic output¶

Definitions of outputs are quite simple, hold optional list of kinds declared for given output.

Definition of Python code¶

Python code is shipped in JSON document with the following fields:

• run_function_code - code of run(...) method of your dynamic block
• run_function_name - name of run function
• init_function_code - optional code for your init function that will assemble step state - it is expected to return dictionary, which will be available for run()function under self._init_results
• init_function_name - name of init function
• imports - list of additional imports (you may only use libraries from your environment, no dependencies will be automatically installed)

How to create run(...) method?¶

You must know the following:

• run(...) function must be defined, as if that was class instance method - with the first argument being self and remaining arguments compatible with dynamic block manifest declared in definition of dynamic block
• you should expect baseline symbols to be provided, including your import statements and the following:

from typing import Any, List, Dict, Set, Optional import supervision as sv import numpy as np import math import time import json import os import requests import cv2 import shapely from inference.core.workflows.execution_engine.entities.base import Batch, WorkflowImageData from inference.core.workflows.prototypes.block import BlockResult

So example function may look like the following (for clarity, we provide here Python code formatted nicely, but you must stringify the code to place it in definition):

def run(self, predictions: sv.Detections, class_x: str, class_y: str) -> BlockResult: bboxes_class_x = predictions[predictions.data["class_name"] == class_x] bboxes_class_y = predictions[predictions.data["class_name"] == class_y] overlap = [] for bbox_x in bboxes_class_x: bbox_x_coords = bbox_x[0] bbox_overlaps = [] for bbox_y in bboxes_class_y: if bbox_y[-1]["detection_id"] == bbox_x[-1]["detection_id"]: continue bbox_y_coords = bbox_y[0] x_min = max(bbox_x_coords[0], bbox_y_coords[0]) y_min = max(bbox_x_coords[1], bbox_y_coords[1]) x_max = min(bbox_x_coords[2], bbox_y_coords[2]) y_max = min(bbox_x_coords[3], bbox_y_coords[3]) # compute the area of intersection rectangle intersection_area = max(0, x_max - x_min + 1) * max(0, y_max - y_min + 1) box_x_area = (bbox_x_coords[2] - bbox_x_coords[0] + 1) * (bbox_x_coords[3] - bbox_x_coords[1] + 1) local_overlap = intersection_area / (box_x_area + 1e-5) bbox_overlaps.append(local_overlap) overlap.append(bbox_overlaps) return {"overlap": overlap}

How to create init(...) method?¶

Init function is supposed to build self._init_results dictionary.

Example:

def my_init() -> Dict[str, Any]: return {"some": "value"}

Usage of Dynamic Python block as step¶

As shown in example Workflow definition, you may simply use the block as if that was normal block exposed through static plugin:

{ "type": "OverlapMeasurement", "name": "overlap_measurement", "predictions": "$steps.model.predictions", "class_x": "dog", "class_y": "dog" }