New Condition block

usage.py

from inference.core.workflows.core_steps.flow_control.new_condition import (
    NewConditionBlock,
)
from inference.core.workflows.core_steps.common.query_language.entities.operations import (
    StatementGroup,
)

# this will be generated in UI and shipped dynamically by EE as run_locally(...) params

condition_statement = StatementGroup.model_validate({
    "type": "StatementGroup",
    "statements": [
        {
            "type": "BinaryStatement",
            "left_operand": {  # gets the detections that meet class_name and confidence criteria and count them
                "type": "DynamicOperand",
                "operand_name": "predictions",
                "operations": 
                    {
                        "type": "DetectionsFilter",
                        "filter_operation": {
                            "type": "StatementGroup",
                            "operator": "and",
                            "statements": [
                                {
                                    "type": "BinaryStatement",
                                    "left_operand": {
                                        "type": "DynamicOperand",
                                        "operations": [
                                            {"type": "ExtractDetectionProperty", "property_name": "class_name"}
                                        ]
                                    },
                                    "comparator": {"type": "in"},
                                    "right_operand": {
                                        "type": "DynamicOperand",
                                        "operand_name": "classes",
                                    }
                                },
                                {
                                    "type": "BinaryStatement",
                                    "left_operand": {
                                        "type": "DynamicOperand",
                                        "operations": [
                                            {"type": "ExtractDetectionProperty", "property_name": "confidence"}
                                        ]
                                    },
                                    "comparator": {"type": "(Number) >="},
                                    "right_operand": {
                                        "type": "DynamicOperand",
                                        "operand_name": "confidence",
                                    }
                                }
                            ]
                        }
                    },
                    {"type": "SequenceLength"}
                ]
            },
            "comparator": { # checks if there is more than X filtered boxes
                "type": "(Number) >=",
            },
            "right_operand": {
                "type": "DynamicOperand",
                "operand_name": "min_objects",
            }
        }
    ]
})


condition_block = NewConditionBlock()

await condition_block.run_locally(
    condition_statement=condition_statement,
    evaluation_parameters={"predictions": detections, "confidence": 0.2, "min_objects": 3, "classes": {"car"}},
    step_if_true="$steps.true",
    step_if_false="$steps.false",
)
>> ([], FlowControl(mode='select_step', context='$steps.false'))


await condition_block.run_locally(
    condition_statement=condition_statement,
    evaluation_parameters={"predictions": detections, "confidence": 0.2, "min_objects": 2, "classes": {"car"}},
    step_if_true="$steps.true",
    step_if_false="$steps.false",
)
>> ([], FlowControl(mode='select_step', context='$steps.true'))