Probe

FluxRender.probes

DataProbe

DataProbe(target_region, target_entity: VectorMathEngine | VectorField | ParticleSystem, measured_property: Property = None)

A measurement instrument that tracks a spatial target and evaluates mathematical properties.

The DataProbe acts as an observer within the simulation. It continuously monitors a specified target region (e.g., an interactive cursor or a moving entity) and queries the underlying mathematical engine. It can track either a specific scalar field property (like divergence or curl) or evaluate the raw 2D vector field at that location. The evaluated results are then dispatched to all registered listener functions via a callback system.

Parameters:

Name	Type	Description	Default
target_region	SpatialRegion	The spatial entity to be tracked. This object must provide a mechanism to retrieve its current coordinates (e.g., a center property or method).	required
target_entity	VectorMathEngine | VectorField | ParticleSystem	The core mathematical engine or entity responsible for parsing and evaluating the vector field logic at the given coordinates.	required
measured_property	Property	The specific mathematical property to measure at the target's location (e.g., Property.DIVERGENCE). If set to None, the probe evaluates the raw underlying vector field, returning a tuple of two values (X and Y vector components). Important Note on Callback Signatures: If measured_property is None, registered callback functions must accept exactly two positional arguments (e.g., def my_callback(vector_x, vector_y):). If measured_property is provided (evaluating to a scalar), registered callback functions must accept exactly one positional argument (e.g., def my_callback(value):). Defaults to None.	None

Example

Creating a DataProbe to track velocity at cursor position:

import FluxRender as fr

# [Initialization of scene and coordinate system]

math_engine = fr.VectorMathEngine(scene=scene, primary_vector_function=lambda x, y: (y, -x))
interactive_cursor = fr.CursorRegion(radius=30) # A source of coordinates that moves with user input

# Define a function that will be called by DataProbe on each frame (if the region is active)
def velocity_callback(velocity_value):
    print(f"Current velocity at cursor: {velocity_value}", end="\r")

probe = fr.DataProbe(
    target_region = interactive_cursor,
    target_entity = math_engine, # You can also use a VectorField or ParticleSystem instance here since they also implement the necessary evaluation interface
    measured_property = fr.Property.VELOCITY
)
probe.add_listener(velocity_callback) # Registers the callback to receive velocity updates at the cursor's position

A comprehensive example of using DataProbe to display velocity:

import FluxRender as fr

scene = fr.create_workspace()

def swirling_vortex(x, y):
    return y, -x

vector_field = fr.VectorField(swirling_vortex)

# To pass data (coordinates) to DataProbe we need a region
cursor = fr.CursorRegion(always_active=True)

# We are creating a DataProbe that will measure velocity
probe = fr.DataProbe(
    target_region=cursor,
    target_entity=vector_field,
    measured_property=fr.Property.VELOCITY
)

# We display velocity using DynamicText
text = fr.DynamicText(text=lambda: f"Velocity: {probe.value}",)


scene.add(vector_field,cursor, probe, text)
scene.run()

Source code in FluxRender/probes.py

def __init__(self, target_region, target_entity: me.VectorMathEngine | en.VectorField | en.ParticleSystem, measured_property: Property = None):
    """
    Args:
        target_region (SpatialRegion): The spatial entity to be tracked. This object must provide
            a mechanism to retrieve its current coordinates (e.g., a `center` property or method).
        target_entity (me.VectorMathEngine | en.VectorField | en.ParticleSystem): The core mathematical engine or entity responsible for parsing and
            evaluating the vector field logic at the given coordinates.
        measured_property (Property, optional): The specific mathematical property to measure
            at the target's location (e.g., Property.DIVERGENCE). If set to None, the probe evaluates
            the raw underlying vector field, returning a tuple of two values (X and Y vector components).

            **Important Note on Callback Signatures:**

            - If `measured_property` is None, registered callback functions must accept exactly
                two positional arguments (e.g., `def my_callback(vector_x, vector_y):`).
            - If `measured_property` is provided (evaluating to a scalar), registered callback
                functions must accept exactly one positional argument (e.g., `def my_callback(value):`).

            Defaults to None.

    Example:
        Creating a DataProbe to track velocity at cursor position:
        ```python
        import FluxRender as fr

        # [Initialization of scene and coordinate system]

        math_engine = fr.VectorMathEngine(scene=scene, primary_vector_function=lambda x, y: (y, -x))
        interactive_cursor = fr.CursorRegion(radius=30) # A source of coordinates that moves with user input

        # Define a function that will be called by DataProbe on each frame (if the region is active)
        def velocity_callback(velocity_value):
            print(f"Current velocity at cursor: {velocity_value}", end="\\r")

        probe = fr.DataProbe(
            target_region = interactive_cursor,
            target_entity = math_engine, # You can also use a VectorField or ParticleSystem instance here since they also implement the necessary evaluation interface
            measured_property = fr.Property.VELOCITY
        )
        probe.add_listener(velocity_callback) # Registers the callback to receive velocity updates at the cursor's position
        ```

        A comprehensive example of using DataProbe to display velocity:
        ```python
        import FluxRender as fr

        scene = fr.create_workspace()

        def swirling_vortex(x, y):
            return y, -x

        vector_field = fr.VectorField(swirling_vortex)

        # To pass data (coordinates) to DataProbe we need a region
        cursor = fr.CursorRegion(always_active=True)

        # We are creating a DataProbe that will measure velocity
        probe = fr.DataProbe(
            target_region=cursor,
            target_entity=vector_field,
            measured_property=fr.Property.VELOCITY
        )

        # We display velocity using DynamicText
        text = fr.DynamicText(text=lambda: f"Velocity: {probe.value}",)


        scene.add(vector_field,cursor, probe, text)
        scene.run()
        ```
    """

    self.target_region = target_region
    self.math_engine = target_entity
    self.measured_property = measured_property

    self.value = 0 if measured_property is not None else (0, 0)  # Initialize value based on property type

    self._callbacks = []

add_listener

add_listener(callback_function)

Adds a function that will be called on every frame update with the latest measurement results. The callback function must have a specific signature based on whether a property is being measured or not:

• If measured_property is None: The callback must accept exactly two positional arguments or *args (e.g., def my_callback(vector_x, vector_y):).
• If measured_property is set: The callback must accept exactly one positional argument (e.g., def my_callback(value):).

Source code in FluxRender/probes.py

def add_listener(self, callback_function):
    """
    Adds a function that will be called on every frame update with the latest measurement results.
    The callback function must have a specific signature based on whether a property is being measured or not:

    - If `measured_property` is None: The callback must accept exactly two positional arguments or *args (e.g., `def my_callback(vector_x, vector_y):`).
    - If `measured_property` is set: The callback must accept exactly one positional argument (e.g., `def my_callback(value):`).
    """

    if not callable(callback_function):
        _fatal_error("Listener must be a callable function.", "TypeError")

    function_name = getattr(callback_function, '__name__', str(callback_function))
    expected_parameters_count = _count_function_parameters(callback_function)

    if self.measured_property is None:
        if expected_parameters_count != 2 and expected_parameters_count != float('inf'):
            _fatal_error(
                f"Callback function '{function_name}' must accept exactly two arguments "
                f"(vector_x, vector_y) or *args when measured_property is None. "
                f"Currently it accepts {expected_parameters_count}.",
                "ValueError"
            )
    else:
        if expected_parameters_count != 1 and expected_parameters_count != float('inf'):
            _fatal_error(
                f"Callback function '{function_name}' must accept exactly one argument "
                f"(value) or *args when measured_property is set. "
                f"Currently it accepts {expected_parameters_count}.",
                "ValueError"
            )

    self._callbacks.append(callback_function)