def pipeline()

in src/mapillary/utils/filter.py [0:0]

• 28 lines of code
• 3 McCabe index (conditional complexity)

def pipeline(data: dict, components: list) -> list:
    """
    A pipeline component that helps with making filtering easier. It provides
    access to different filtering mechanism by simplying letting users
    pass in what filter they want to apply, and the arguments for that filter

    :param data: The GeoJSON to be filtered
    :type data: dict

    :param components: The list of filters to apply
    :type components: list

    :return: The filtered feature list
    :rtype: list

    Usage::

        >>> # assume variables 'data', 'kwargs'
        >>> pipeline(
        ...     data=data,
        ...     components=[
        ...         {"filter": "image_type", "tile": kwargs["image_type"]}
        ...         if "image_type" in kwargs
        ...         else {},
        ...         {"filter": "organization_id", "organization_ids": kwargs["org_id"]}
        ...         if "org_id" in kwargs
        ...         else {},
        ...         {
        ...             "filter": "haversine_dist",
        ...             "radius": kwargs["radius"],
        ...             "coords": [longitude, latitude],
        ...         }
        ...         if "radius" in kwargs
        ...         else 1000
        ...     ]
        ... )
    """

    # Python treats dict objects as passed reference, thus
    # in order to not modify the previous state, we make a local copy
    __data = data.copy()["features"]

    # A mapping of different filters possible
    function_mappings = {
        "filter_values": filter_values,
        "max_captured_at": max_captured_at,
        "min_captured_at": min_captured_at,
        "haversine_dist": haversine_dist,
        "image_type": image_type,
        "organization_id": organization_id,
        "features_in_bounding_box": features_in_bounding_box,
        "existed_at": existed_at,
        "existed_before": existed_before,
        "sequence_id": sequence_id,
        "compass_angle": compass_angle,
        "hits_by_look_at": hits_by_look_at,
        "in_shape": in_shape,
        # Simply add the mapping of a new function,
        # nothing else will really need to changed
    }

    # Going through each of the components
    for component in components:

        # If component is simply empty, continue to next
        # iteration
        if component == {}:
            continue

        # Send to pipeline component, return data to `__data`
        __data = pipeline_component(
            # Map function respectively using the function_mappings dictionary
            func=function_mappings[f'{component["filter"]}'],
            # Send over the data
            data=__data,
            # Specify the message on the exception thrown
            exception_message=f'[pipeline - {component["filter"]}] Filter not applied, '
            "exception thrown",
            # Except the filter name, select the rest as args
            args=tuple(list(component.values())[1:]),
        )

    # Return the data
    return __data