meerschaum.utils.pipes

  1#! /usr/bin/env python3
  2# vim:fenc=utf-8
  3
  4"""
  5Define utilities for working with pipes.
  6"""
  7
  8from __future__ import annotations
  9
 10from typing import Any, Dict, Callable
 11import re
 12import ast
 13import copy
 14
 15from meerschaum.utils.typing import PipesDict, Optional
 16import meerschaum as mrsm
 17from meerschaum import _core
 18
 19
 20def get_pipe_from_value(
 21    value: str | dict[str, str],
 22    _pipe: mrsm.Pipe | None = None,
 23) -> mrsm.Pipe | None:
 24    """
 25    Given an input value (string or dictionary), return the corresponding pipe.
 26
 27    Parameters
 28    ----------
 29    value: str | dict[str, str]
 30        The definition keys for the pipe to be constructed.
 31        This may either be the literal constructor string or a dictionary of constructor kwargs.
 32
 33    _pipe: mrsm.Pipe | None, default None
 34        If provided, use the instance connector for this reference pipe
 35        if an explicit instance connector is undefined.
 36
 37    Returns
 38    -------
 39    A `Pipe` corresponding to the given keys.
 40    """
 41    from meerschaum.utils.warnings import warn
 42    if isinstance(value, mrsm.Pipe):
 43        return value
 44
 45    if isinstance(value, str):
 46        return get_pipe_from_string(value, _pipe=_pipe)
 47
 48    if not isinstance(value, dict):
 49        raise ValueError("Expecting kwargs for a Pipe.")
 50
 51    if (
 52        _pipe is not None
 53        and not (
 54            'instance' in value
 55            or 'mrsm_instance' in value
 56            or 'instance_keys' in value
 57        )
 58    ):
 59        value = {**value}
 60        value['instance'] = _pipe.instance_keys
 61
 62    try:
 63        pipe = mrsm.Pipe(**value)
 64    except Exception as e:
 65        warn(f"Failed build pipe from value '{value}':\n{e}")
 66        return None
 67
 68    return pipe
 69
 70
 71def get_pipe_from_string(
 72    string: str,
 73    _pipe: mrsm.Pipe | None = None,
 74) -> mrsm.Pipe | None:
 75    """
 76    If a string is equal to a `Pipe` constructor, return the pipe.
 77
 78    Parameters
 79    ----------
 80    string: str
 81        The string containing the constructor syntax.
 82        Must be an exact match.
 83
 84    _pipe: mrsm.Pipe | None, default None
 85        If provided, use the instance connector for this reference pipe
 86        if an explicit instance connector is undefined.
 87
 88    Returns
 89    -------
 90    A `Pipe` corresponding the defined pipe in the input string.
 91    
 92    Examples
 93    --------
 94    >>> get_pipe_from_string('Pipe("foo", "bar")')
 95    Pipe('foo', 'bar')
 96    >>> get_pipe_from_string('mrsm.Pipe("spam", "eggs", instance="sql:local")')
 97    Pipe('spam', 'eggs', instance='sql:local')
 98    """
 99    from meerschaum.utils.warnings import warn
100    from meerschaum.utils.misc import parse_arguments_str
101    pattern = r'(?:mrsm\.)?Pipe\((.*?)\)'
102    matches = list(re.finditer(pattern, string))
103    if not matches:
104        return None
105
106    try:
107        match = matches[0]
108        args_str = match.group(1)
109        args, kwargs = parse_arguments_str(args_str)
110        if (
111            _pipe is not None
112            and not (
113                'instance' in kwargs
114                or 'mrsm_instance' in kwargs
115                or 'instance_keys' in kwargs
116            )
117        ):
118            kwargs['instance'] = _pipe.instance_keys
119        pipe = mrsm.Pipe(*args, **kwargs)
120    except Exception as e:
121        warn(f"Failed to build pipe from string '{string}': \n{e}")
122        return None
123
124    return pipe
125
126
127def evaluate_pipe_access_chain(access_chain: str, pipe: mrsm.Pipe, _pipe=None):
128    """
129    Safely evaluate the access chain on a Pipe.
130    """
131    expr = f"pipe{access_chain}"
132    if pipe == _pipe and access_chain.lstrip('.').startswith('parameters'):
133        parameters_access_chain = access_chain.lstrip('.').split('parameters', maxsplit=1)[-1]
134        expr = f"pipe.attributes['parameters']{parameters_access_chain}"
135    tree = ast.parse(expr, mode='eval')
136
137    def _eval(node, context):
138        if isinstance(node, ast.Expression):
139            return _eval(node.body, context)
140
141        elif isinstance(node, ast.Name):
142            if node.id == "pipe":
143                return context
144            raise ValueError(f"Unknown variable: {node.id}")
145
146        elif isinstance(node, ast.Attribute):
147            value = _eval(node.value, context)
148            return getattr(value, node.attr)
149
150        elif isinstance(node, ast.Subscript):
151            value = _eval(node.value, context)
152            key = _eval(node.slice, context) if isinstance(node.slice, ast.Index) else _eval(node.slice, context)
153            return value[key]
154
155        elif isinstance(node, ast.Constant):  # Python 3.8+
156            return node.value
157
158        elif isinstance(node, ast.Str):  # Older Python
159            return node.s
160
161        elif isinstance(node, ast.Index):  # Older Python AST style
162            return _eval(node.value, context)
163
164        else:
165            raise TypeError(f"Unsupported AST node: {ast.dump(node)}")
166
167    return _eval(tree, pipe)
168
169
170
171def replace_pipes_syntax(text: str, _pipe=None) -> Any:
172    """
173    Parse a string containing the `{{ Pipe() }}` syntax.
174    """
175    _rust = _core.replace_pipes_syntax
176    return _rust(text, _pipe=_pipe)
177
178
179def replace_pipes_in_dict(
180    pipes: Optional[PipesDict] = None,
181    func: Callable[[Any], Any] = str,
182    debug: bool = False,
183    **kw
184) -> PipesDict:
185    """
186    Replace the Pipes in a Pipes dict with the result of another function.
187
188    Parameters
189    ----------
190    pipes: Optional[PipesDict], default None
191        The pipes dict to be processed.
192
193    func: Callable[[Any], Any], default str
194        The function to be applied to every pipe.
195        Defaults to the string constructor.
196
197    debug: bool, default False
198        Verbosity toggle.
199
200    Returns
201    -------
202    A dictionary where every pipe is replaced with the output of a function.
203
204    """
205    def change_dict(d: Dict[Any, Any]) -> None:
206        for k, v in d.items():
207            if isinstance(v, dict):
208                change_dict(v)
209            elif isinstance(v, list):
210                d[k] = [func(i) for i in v]
211            elif isinstance(v, tuple):
212                d[k] = tuple([func(i) for i in v])
213            else:
214                d[k] = func(v)
215
216    if pipes is None:
217        from meerschaum import get_pipes
218        pipes = get_pipes(debug=debug, **kw)
219
220    result = copy.deepcopy(pipes)
221    change_dict(result)
222    return result
223
224
225def is_pipe_registered(
226    pipe: mrsm.Pipe,
227    pipes: PipesDict,
228    debug: bool = False
229) -> bool:
230    """
231    Check if a Pipe is inside the pipes dictionary.
232
233    Parameters
234    ----------
235    pipe: meerschaum.Pipe
236        The pipe to see if it's in the dictionary.
237
238    pipes: PipesDict
239        The dictionary to search inside.
240
241    debug: bool, default False
242        Verbosity toggle.
243
244    Returns
245    -------
246    A bool indicating whether the pipe is inside the dictionary.
247    """
248    ck, mk, lk = pipe.connector_keys, pipe.metric_key, pipe.location_key
249    try:
250        return ck in pipes and mk in pipes[ck] and lk in pipes[ck][mk]
251    except KeyError:
252        return False
253
254
255def pipes_dict_from_list(pipes_list: list[mrsm.Pipe]) -> PipesDict:
256    """
257    Return a PipesDict from a list of pipes.
258    Note that all pipes must share the same instance connector.
259
260    Examples
261    --------
262    >>> pipes_list = [mrsm.Pipe('a', 'b'), mrsm.Pipe('a', 'b', 'c')]
263    >>> pipes_dict_from_list(pipes)
264    {
265        'a': {
266            'b': {
267                None: Pipe('a', 'b')
268                'c': Pipe('a', 'b', 'c')
269            }
270        }
271    }
272    """
273    from meerschaum.utils.warnings import warn
274    from meerschaum.utils.misc import items_str
275    if not pipes_list:
276        return {}
277
278    instance_keys = pipes_list[0].instance_keys
279    same_pipes = [pipe for pipe in pipes_list if pipe.instance_keys == instance_keys]
280    diff_pipes = [pipe for pipe in pipes_list if pipe.instance_keys != instance_keys]
281    if diff_pipes:
282        warn(
283            f"Skipping pipes with instance keys different from '{instance_keys}':\n"
284            f"{items_str(diff_pipes, quotes=False)}"
285        )
286
287    pipes_dict: PipesDict = {}
288    for pipe in same_pipes:
289        if pipe.connector_keys not in pipes_dict:
290            pipes_dict[pipe.connector_keys] = {}
291        if pipe.metric_key not in pipes_dict[pipe.connector_key]:
292            pipes_dict[pipe.connector_keys][pipe.metric_key] = {}
293        pipes_dict[pipe.connector_keys][pipe.metric_key][pipe.location_key] = pipe
294
295    return pipes_dict
296
297
298def flatten_pipes_dict(pipes_dict: PipesDict) -> list[mrsm.Pipe]:
299    """
300    Convert the standard pipes dictionary into a list.
301
302    Parameters
303    ----------
304    pipes_dict: PipesDict
305        The pipes dictionary to be flattened.
306
307    Returns
308    -------
309    A list of `Pipe` objects.
310
311    """
312    pipes_list = []
313    for ck in pipes_dict.values():
314        for mk in ck.values():
315            pipes_list.extend(list(mk.values()))
316    return pipes_list