cortex_loom.flow_builder

FlowBuilder(api_token: str)

FlowBuilder helps validate custom functions meant for publishing in Cortex and provides an interface for the publishing API.

Parameters:

Name	Type	Description	Default
api_token	str	API token to validate the Flow publishing requests with.	required

Examples:

Basic publishing of a simple function

>>> def extract_video_metadata(
...     video: VideoObject,
... ) -> float:
...     '''Reads metadata of the video (duration and frame dimensions).'''
...     return video.duration, video.height, video.width
...
>>> builder = FlowBuilder(api_token="your-long-api-token-here")
>>> response = builder.publish(
...     extract_video_metadata,
...     flow_display_name="Video Metadata Extraction",
...     flow_description="Extracts metadata of the video.",
...     output_display_name="Video Metadata",
... )

Validate function signature & serialisation before publishing

>>> serialised_func, param_spec = builder.build(extract_video_metadata)
>>> print(param_spec)
{'video': {
    'data_type': 'VideoObject',
    'provider_type': 'cortex_loom.types.video_object.VideoObjectProvider',
    'provider_parameters': { # Filled with Provider's defaults
        "stream_index": 0,
        "pixel_format": "rgb24",
        "rotation": None,
        "height": None,
        "width": None,
    }
}}

Publishing with custom VideoObjectProvider parameters (force RGB format, resize to 640×360, rotate 90° clockwise)

>>> # Tell the platform to load input videos with specific transformations
>>> custom_params = {
...     "video": {                       # name of the function parameter
...         "width": 640,
...         "height": 360,
...         "rotation": 90,              # clockwise
...     }
... }
...
>>> response = builder.publish(
...     extract_video_metadata,
...     custom_provider_parameters=custom_params,
...     ...,
... )
>>>
>>> # Or just validate without publishing
>>> serialised, spec = builder.build(
...     extract_upper_half,
...     custom_provider_parameters=custom_params,
... )
>>> print(spec["video"]["provider_parameters"])
{
    'stream_index': 0,
    'pixel_format': 'rgb24',
    'width': 640,
    'height': 360,
    'rotation': 90
}

build(func: Callable, custom_provider_parameters: Dict[str, Dict[str, Any]] = {}) -> Tuple[BytesIO, Dict[str, dict]]

Validates the signature of the function and prepares its input specification. If successful, returns the serialised function and parameter details. Can be used to validate the function before publishing.

Parameters:

Name	Type	Description	Default
func	Callable	Function to be serialised. Its parameters need to be strictly type-hinted and the output has to be JSON-serialisable.	required
custom_provider_parameters	Dict[str, Dict[str, Any]]	Mapping of function input names to a dictionary of the input's DataObjectProvider parameters. Used to overwrite default parameter values of the Providers.	{}

Returns:

Type	Description
BytesIO	Buffer containing the serialised func code.
Dict[str, dict]	Mapping of function parameter names to their details including type name and Provider parameters.

publish(func: Callable, custom_provider_parameters: Dict[str, Dict[str, Any]] = {}, flow_display_name: Optional[str] = None, flow_description: Optional[str] = None, output_display_name: Optional[str] = None, output_description: Optional[str] = None) -> Dict[str, Any]

Publishes the given function as a custom Cortex Flow.

Parameters:

Name	Type	Description	Default
func	Callable	Function to be published. Its parameters need to be strictly type-hinted and the output has to be JSON-serialisable. Once published, it will be available for use in Kelvin Cortex.	required
custom_provider_parameters	Dict[str, Dict[str, Any]]	Mapping of function input names to a dictionary of the input's DataObjectProvider parameters. Used to overwrite default parameter values of the Providers.	{}
flow_display_name	Optional[str]	Name of the function to be displayed in Cortex. If None, a name will be generated.	None
flow_description	Optional[str]	Optional description of the Flow shown to users in Cortex.	None
output_display_name	Optional[str]	Name of the output of the function. If None, a name will be generated.	None
output_description	Optional[str]	Optional description of the Flow's output visible to Cortex users.	None

Returns:

Type	Description
Dict[str, Any]	Response of the publishing API.