meerschaum.utils.interactive

 1#! /usr/bin/env python3
 2# -*- coding: utf-8 -*-
 3# vim:fenc=utf-8
 4
 5"""
 6Interactive "wizards" to guide the user.
 7"""
 8
 9from __future__ import annotations
10
11import meerschaum as mrsm
12from meerschaum.utils.typing import List
13
14def select_pipes(
15        yes: bool = False,
16        force: bool = False,
17        debug: bool = False,
18    ) -> List[mrsm.Pipe]:
19    """Prompt the user for the keys to identify a list of pipes."""
20    from meerschaum.utils.misc import get_connector_labels
21    from meerschaum.utils.prompt import prompt, choose, yes_no
22    from meerschaum.utils import get_pipes
23    from meerschaum.utils.formatting._shell import clear_screen
24    from meerschaum.utils.formatting import pprint_pipes
25    from meerschaum.config import get_config
26    from meerschaum.utils.warnings import warn
27    from meerschaum.utils.misc import flatten_pipes_dict
28    from meerschaum.connectors import instance_types
29    clear_screen(debug=debug)
30    while True:
31        instance = choose(
32            "On which instance are the pipes stored?",
33            get_connector_labels(*instance_types),
34            numeric = True,
35            default = get_config('meerschaum', 'instance'),
36        )
37        pipes_dict = get_pipes(instance=instance, debug=debug)
38        if pipes_dict:
39            break
40        warn(f"There are no pipes registered on instance '{instance}'.", stack=False)
41        print("    Please choose another instance.")
42
43    conn_keys = sorted(list(pipes_dict.keys()))
44    clear_screen(debug=debug)
45    chosen_conn_keys = choose(
46        "What are the connectors for the pipes?",
47        conn_keys,
48        numeric = True,
49        multiple = True,
50        default = (conn_keys[0] if len(conn_keys) == 1 else None)
51    )
52
53    pipes_dict = get_pipes(chosen_conn_keys, instance=instance, debug=debug)
54    metric_keys = []
55    for ck, metrics in pipes_dict.items():
56        for mk in metrics:
57            metric_keys.append(mk)
58    metric_keys = sorted(metric_keys)
59    clear_screen(debug=debug)
60    chosen_metric_keys = choose(
61        "What are the metrics for the pipes?",
62        metric_keys,
63        numeric = True,
64        multiple = True,
65        default = (metric_keys[0] if len(metric_keys) == 1 else None)
66    )
67   
68    pipes_dict = get_pipes(chosen_conn_keys, chosen_metric_keys, instance=instance, debug=debug)
69    location_keys = []
70    for ck, metrics in pipes_dict.items():
71        for mk, locations in metrics.items():
72            for lk, p in locations.items():
73                location_keys.append(str(lk))
74    location_keys = sorted(location_keys)
75    clear_screen(debug=debug)
76    chosen_location_keys = [(lk if lk != 'None' else None) for lk in choose(
77        "What are the locations for the pipes?",
78        location_keys,
79        numeric = True,
80        multiple = True,
81        default = (location_keys[0] if len(location_keys) == 1 else None)
82    )]
83
84    pipes_dict = get_pipes(
85        chosen_conn_keys,
86        chosen_metric_keys,
87        chosen_location_keys,
88        instance = instance,
89        debug = debug,
90    )
91    clear_screen(debug=debug)
92    pprint_pipes(pipes_dict)
93    if force or yes_no("Choose these pipes?", yes=yes):
94        return flatten_pipes_dict(pipes_dict)
95    return select_pipes(yes=yes, force=force, debug=debug)