handler.py

"""
    Auxiliary handler methods for data summary extraction
"""
from typing import Any, Callable, Dict, List, Sequence, Tuple

import networkx as nx
from visions import VisionsTypeset


def compose(functions: Sequence[Callable[..., Any]]) -> Callable[..., Tuple[Any, ...]]:
    """
    Compose a sequence of functions.

    Each function in the sequence should accept the arguments passed to the composed
    function and return either a single value or a tuple of values.
    
    :param functions: sequence of functions
    :return: combined function applying all functions in order.
    """

    def composed_function(*args: Any) -> Tuple[Any, ...]:
        result: Tuple[Any, ...] = args
        for func in functions:
            result = func(*result)
            # Ensure result is always a tuple for consistent unpacking
            if not isinstance(result, tuple):
                result = (result,)
        return result

    return composed_function


class Handler:
    """A generic handler

    Allows any custom mapping between data types and functions
    """

    def __init__(
        self,
        mapping: Dict[str, List[Callable[..., Any]]],
        typeset: VisionsTypeset,
        *args: Any,
        **kwargs: Any
    ):
        self.mapping = mapping
        self.typeset = typeset
        self._complete_dag()

    def _complete_dag(self) -> None:
        for from_type, to_type in nx.topological_sort(
            nx.line_graph(self.typeset.base_graph)
        ):
            self.mapping[str(to_type)] = (
                self.mapping[str(from_type)] + self.mapping[str(to_type)]
            )

    def handle(self, dtype: str, *args: Any, **kwargs: Any) -> Any:
        """
        Execute the handler chain for the given dtype.
        
        :param dtype: The data type to handle
        :param args: Arguments to pass to the handler chain
        :return: The last element of the result tuple from the handler chain
        """
        funcs = self.mapping.get(dtype, [])
        op = compose(funcs)
        summary = op(*args)[-1]
        return summary


def get_render_map() -> Dict[str, Callable[..., Any]]:
    import ydata_profiling.report.structure.variables as render_algorithms

    render_map: Dict[str, Callable[..., Any]] = {
        "Boolean": render_algorithms.render_boolean,
        "Numeric": render_algorithms.render_real,
        "Complex": render_algorithms.render_complex,
        "Text": render_algorithms.render_text,
        "DateTime": render_algorithms.render_date,
        "Categorical": render_algorithms.render_categorical,
        "URL": render_algorithms.render_url,
        "Path": render_algorithms.render_path,
        "File": render_algorithms.render_file,
        "Image": render_algorithms.render_image,
        "Unsupported": render_algorithms.render_generic,
        "TimeSeries": render_algorithms.render_timeseries,
    }

    return render_map