|
1 | 1 | """ |
2 | 2 | Auxiliary handler methods for data summary extraction |
3 | 3 | """ |
4 | | -from typing import Any, Callable, Dict, List, Sequence, Tuple, TypeVar, cast |
| 4 | +from typing import Any, Callable, Dict, List, Sequence, Tuple, Union |
5 | 5 |
|
6 | 6 | import networkx as nx |
7 | 7 | from visions import VisionsTypeset |
8 | 8 |
|
9 | | -T = TypeVar("T") |
10 | | -SummaryFunction = Callable[..., Tuple[Any, ...]] |
11 | 9 |
|
12 | | - |
13 | | -def compose(functions: Sequence[SummaryFunction]) -> SummaryFunction: |
| 10 | +def compose(functions: Sequence[Callable]) -> Callable: |
14 | 11 | """ |
15 | 12 | Compose a sequence of functions. |
16 | 13 |
|
17 | | - :param functions: sequence of functions |
18 | | - :return: combined function applying all functions in order. |
| 14 | + Each function in the sequence receives the result of the previous function. |
| 15 | + Functions are expected to accept and return tuples for proper chaining. |
| 16 | +
|
| 17 | + :param functions: sequence of functions that accept and return tuples |
| 18 | + :return: combined function applying all functions in order |
19 | 19 | """ |
20 | 20 |
|
21 | 21 | def composed_function(*args: Any) -> Tuple[Any, ...]: |
22 | | - result: Tuple[Any, ...] = args |
| 22 | + result: Union[Tuple[Any, ...], Any] = args |
23 | 23 | for func in functions: |
24 | | - step_result = func(*result) |
25 | | - if not isinstance(step_result, tuple): |
26 | | - result = (step_result,) |
| 24 | + if isinstance(result, tuple): |
| 25 | + result = func(*result) |
27 | 26 | else: |
28 | | - result = step_result |
29 | | - return result |
| 27 | + result = func(result) |
| 28 | + if isinstance(result, tuple): |
| 29 | + return result |
| 30 | + return (result,) |
30 | 31 |
|
31 | 32 | return composed_function |
32 | 33 |
|
33 | 34 |
|
34 | 35 | class Handler: |
35 | 36 | """A generic handler |
36 | 37 |
|
37 | | - Allows any custom mapping between data types and functions |
| 38 | + Allows any custom mapping between data types and functions. |
| 39 | + Functions are composed based on the type hierarchy defined in the typeset. |
38 | 40 | """ |
39 | 41 |
|
40 | 42 | def __init__( |
41 | 43 | self, |
42 | | - mapping: Dict[str, List[SummaryFunction]], |
| 44 | + mapping: Dict[str, List[Callable]], |
43 | 45 | typeset: VisionsTypeset, |
44 | 46 | *args: Any, |
45 | | - **kwargs: Any, |
46 | | - ) -> None: |
47 | | - self.mapping: Dict[str, List[SummaryFunction]] = mapping |
| 47 | + **kwargs: Any |
| 48 | + ): |
| 49 | + self.mapping = mapping |
48 | 50 | self.typeset = typeset |
49 | 51 | self._complete_dag() |
50 | 52 |
|
51 | 53 | def _complete_dag(self) -> None: |
52 | 54 | for from_type, to_type in nx.topological_sort( |
53 | 55 | nx.line_graph(self.typeset.base_graph) |
54 | 56 | ): |
55 | | - from_type_str = str(from_type) |
56 | | - to_type_str = str(to_type) |
57 | | - |
58 | | - if from_type_str not in self.mapping: |
59 | | - continue |
60 | | - |
61 | | - if to_type_str in self.mapping: |
62 | | - self.mapping[to_type_str] = ( |
63 | | - self.mapping[from_type_str] + self.mapping[to_type_str] |
64 | | - ) |
65 | | - else: |
66 | | - self.mapping[to_type_str] = self.mapping[from_type_str].copy() |
| 57 | + from_key = str(from_type) |
| 58 | + to_key = str(to_type) |
| 59 | + self.mapping[to_key] = self.mapping.get(from_key, []) + self.mapping.get( |
| 60 | + to_key, [] |
| 61 | + ) |
67 | 62 |
|
68 | 63 | def handle(self, dtype: str, *args: Any, **kwargs: Any) -> Dict[str, Any]: |
69 | 64 | """ |
70 | | - Returns: |
71 | | - object: a tuple containing the config, the dataset series and the summary extracted |
| 65 | + Execute the handler chain for the given data type. |
| 66 | +
|
| 67 | + :param dtype: the data type to handle |
| 68 | + :param args: arguments to pass to the handler functions |
| 69 | + :param kwargs: keyword arguments (currently unused but reserved for extensibility) |
| 70 | + :return: a dictionary containing the summary extracted from the data |
72 | 71 | """ |
73 | 72 | funcs = self.mapping.get(dtype, []) |
74 | 73 | op = compose(funcs) |
75 | 74 | result = op(*args) |
76 | | - return cast(Dict[str, Any], result[-1]) |
| 75 | + if result: |
| 76 | + return result[-1] if isinstance(result[-1], dict) else {} |
| 77 | + return {} |
77 | 78 |
|
78 | 79 |
|
79 | 80 | def get_render_map() -> Dict[str, Callable]: |
|
0 commit comments