hoops_ai.flowmanager.flow_builder

HOOPS CAD Flow Builder with ETL-Based Task Decorators

This module provides a flow builder interface for customizing CAD processing workflows using the ETL (Extract, Transform, Load) pattern with type-safe Python decorators.

KEY FEATURES:

  1. ETL-Based Task Decorators: - @flowtask.extract: Data extraction phase (e.g., gathering CAD files) - @flowtask.transform: Data transformation phase (e.g., encoding CAD data) - @flowtask.custom: Flexible custom tasks for any processing

  2. Simplified Flow Creation: - hoops_ai.create_flow(): Create flows with direct parameters - Simple max_workers parameter for parallel execution - Automatic dataset export integration

  3. Automatic Function Injection: - Decorators inject user functions into task execution framework - No need to implement new ParallelTask classes - Task names are user-defined, not hard-coded

  4. Execution Mode Control: - parallel_execution parameter on decorators for explicit control - Automatic sequential mode when max_workers <= 1 - Per-task execution mode override

USAGE EXAMPLE:

from hoops_ai.flowmanager import flowtask import hoops_ai

@flowtask.extract(

name=”gather_cad_files”, inputs=[“cad_datasources”], outputs=[“cad_dataset”], parallel_execution=True

) def my_gather_function(source: str) -> List[str]:

return glob.glob(f”{source}/*.step”)

@flowtask.transform(

name=”encode_cad”, inputs=[“cad_file”, “cad_loader”, “storage”], outputs=[“face_count”, “edge_count”], parallel_execution=True

) def my_encode_function(cad_file, cad_loader, storage):

# transformation logic return face_count, edge_count

# Create flow with simplified parameters my_flow = hoops_ai.create_flow(

name=”my_flow”, tasks=[my_gather_function, my_encode_function], flows_outputdir=”./output”, max_workers=4, debug=False

)

# Execute flow flow_output, summary, flow_file = my_flow.process(inputs={“cad_datasources”: [“/path/to/cad”]})

Functions

create_flow(name, tasks, flows_outputdir[, ...])

Module-level flow creation function with simplified parameters.

list_registered_tasks()

Returns a dictionary of all registered tasks.

print_task_registry()

Prints information about all registered tasks.

validate_function_signature(func, ...)

Validates that a function has the expected signature for the task type.

validate_tasks()

Validates all registered tasks and returns validation results.

Classes

ParallelExecutor(max_workers)

Simplified parallel executor configuration

StorageType(value[, names, module, ...])

Storage type enumeration for data storage selection

TaskDecorators()

ETL-based task decorators for flow orchestration.
class hoops_ai.flowmanager.flow_builder.ParallelExecutor(max_workers)

Bases: object

Simplified parallel executor configuration

Parameters:

max_workers (int)

class hoops_ai.flowmanager.flow_builder.StorageType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Storage type enumeration for data storage selection

JSON = 'json'
OPTIMIZED = 'optimized'
class hoops_ai.flowmanager.flow_builder.TaskDecorators

Bases: object

ETL-based task decorators for flow orchestration.

Supports ETL (Extract, Transform, Load) pattern: - @flowtask.extract: Data extraction phase (e.g., file gathering) - @flowtask.transform: Data transformation phase (e.g., encoding) - @flowtask.custom: Flexible custom tasks for any processing

custom(name=None, inputs=None, outputs=None, parallel_execution=True)

Decorator for custom generic tasks with flexible signatures.

Parameters:

  • name (str) – User-defined task name (defaults to function name)

  • inputs (list) – Optional input parameter names

  • outputs (list) – Optional output parameter names

  • parallel_execution (bool) – Whether to execute in parallel (default: True)

extract(name=None, inputs=None, outputs=None, parallel_execution=True)

ETL Extract phase decorator - for data extraction tasks.

Parameters:

  • name (str) – User-defined task name (defaults to function name)

  • inputs (list) – List of input parameter names

  • outputs (list) – List of output parameter names

  • parallel_execution (bool) – Whether to execute in parallel

transform(name=None, inputs=None, outputs=None, parallel_execution=True)

ETL Transform phase decorator - for data transformation tasks.

Parameters:

  • name (str) – User-defined task name (defaults to function name)

  • inputs (list) – List of input parameter names

  • outputs (list) – List of output parameter names

  • parallel_execution (bool) – Whether to execute in parallel

hoops_ai.flowmanager.flow_builder.create_flow(name, tasks, flows_outputdir, max_workers=None, ml_task=None, debug=False, auto_dataset_export=True)

Module-level flow creation function with simplified parameters.

Parameters:

  • name (str) – Name of the flow

  • tasks (List[Any]) – List of decorated functions to execute

  • flows_outputdir (str) – Output directory for flow results

  • max_workers (int) – Number of parallel workers (None = auto-detect CPU count)

  • ml_task (str) – Optional ML task description

  • debug (bool) – Enable debug logging

  • auto_dataset_export (bool) – Automatically inject dataset export task

Returns:

Configured Flow instance

Example

>>> import hoops_ai
>>> from hoops_ai.flowmanager import flowtask
>>>
>>> @flowtask.transform(name="encode", parallel_execution=True)
>>> def my_encoder(cad_file, cad_loader, storage):
>>>     # transformation logic
>>>     return face_count, edge_count
>>>
>>> flow = hoops_ai.create_flow(
>>>     name="my_flow",
>>>     tasks=[my_encoder],
>>>     flows_outputdir="./output",
>>>     max_workers=4
>>> )
hoops_ai.flowmanager.flow_builder.list_registered_tasks()

Returns a dictionary of all registered tasks.

Returns:

Dictionary with task types as keys and lists of task names as values

Return type:

Dict[str, Any]

hoops_ai.flowmanager.flow_builder.print_task_registry()

Prints information about all registered tasks.

hoops_ai.flowmanager.flow_builder.validate_function_signature(func, expected_params, task_type)

Validates that a function has the expected signature for the task type.

Parameters:

  • func (Callable) – The function to validate

  • expected_params (List[str]) – List of expected parameter names

  • task_type (str) – Type of task for error messages

Returns:

True if signature is valid

Raises:

ValueError – If signature is invalid

Return type:

bool

hoops_ai.flowmanager.flow_builder.validate_tasks()

Validates all registered tasks and returns validation results.

Returns:

Dictionary with validation results including valid_tasks, invalid_tasks, and warnings

Return type:

Dict[str, Any]