summaryrefslogtreecommitdiff
path: root/dict_utils.py
diff options
context:
space:
mode:
authorScott Gasch <[email protected]>2021-10-23 23:49:08 -0700
committerScott Gasch <[email protected]>2021-10-23 23:49:08 -0700
commitd08bad64a6884f25d28a2c38c6cd1c87b4335188 (patch)
tree9d0e4381b4da1a543acf907b769d1d4d88a3377e /dict_utils.py
parent351e77c767c9084aa486eedbdc9902c635b06261 (diff)
Adds site_config; adds Tuya lights. Bugfixes.
Diffstat (limited to 'dict_utils.py')
-rw-r--r--dict_utils.py71
1 files changed, 61 insertions, 10 deletions
diff --git a/dict_utils.py b/dict_utils.py
index 92fd1e0..7b0edb5 100644
--- a/dict_utils.py
+++ b/dict_utils.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
from itertools import islice
-from typing import Any, Callable, Dict, Iterator, Tuple
+from typing import Any, Callable, Dict, Iterator, List, Tuple
def init_or_inc(
@@ -44,16 +44,24 @@ def shard(d: Dict[Any, Any], size: int) -> Iterator[Dict[Any, Any]]:
yield {key: value for (key, value) in islice(items, x, x + size)}
-def coalesce_by_creating_list(key, v1, v2):
+def coalesce_by_creating_list(key, new_value, old_value):
from list_utils import flatten
- return flatten([v1, v2])
+ return flatten([new_value, old_value])
-def coalesce_by_creating_set(key, v1, v2):
- return set(coalesce_by_creating_list(key, v1, v2))
+def coalesce_by_creating_set(key, new_value, old_value):
+ return set(coalesce_by_creating_list(key, new_value, old_value))
-def raise_on_duplicated_keys(key, v1, v2):
+def coalesce_last_write_wins(key, new_value, old_value):
+ return new_value
+
+
+def coalesce_first_write_wins(key, new_value, old_value):
+ return old_value
+
+
+def raise_on_duplicated_keys(key, new_value, old_value):
raise Exception(f'Key {key} is duplicated in more than one input dict.')
@@ -62,10 +70,13 @@ def coalesce(
*,
aggregation_function: Callable[[Any, Any], Any] = coalesce_by_creating_list
) -> Dict[Any, Any]:
- """Merge N dicts into one dict containing the union of all keys/values in
- the input dicts. When keys collide, apply the aggregation_function which,
- by default, creates a list of values. See also coalesce_by_creating_set or
- provide a user defined aggregation_function.
+ """Merge N dicts into one dict containing the union of all keys /
+ values in the input dicts. When keys collide, apply the
+ aggregation_function which, by default, creates a list of values.
+ See also several other alternative functions for coalescing values
+ (coalesce_by_creating_set, coalesce_first_write_wins,
+ coalesce_last_write_wins, raise_on_duplicated_keys) or provide a
+ custom helper function.
>>> a = {'a': 1, 'b': 2}
>>> b = {'b': 1, 'c': 2, 'd': 3}
@@ -73,6 +84,14 @@ def coalesce(
>>> coalesce([a, b, c])
{'a': 1, 'b': [1, 2], 'c': [1, 2], 'd': [2, 3]}
+ >>> coalesce([a, b, c], aggregation_function=coalesce_last_write_wins)
+ {'a': 1, 'b': 1, 'c': 1, 'd': 2}
+
+ >>> coalesce([a, b, c], aggregation_function=raise_on_duplicated_keys)
+ Traceback (most recent call last):
+ ...
+ Exception: Key b is duplicated in more than one input dict.
+
"""
out: Dict[Any, Any] = {}
for d in inputs:
@@ -177,6 +196,38 @@ def min_key(d: Dict[Any, Any]) -> Any:
return min(d.keys())
+def parallel_lists_to_dict(keys: List[Any], values: List[Any]) -> Dict[Any, Any]:
+ """Given two parallel lists (keys and values), create and return
+ a dict.
+
+ >>> k = ['name', 'phone', 'address', 'zip']
+ >>> v = ['scott', '555-1212', '123 main st.', '12345']
+ >>> parallel_lists_to_dict(k, v)
+ {'name': 'scott', 'phone': '555-1212', 'address': '123 main st.', 'zip': '12345'}
+
+ """
+ if len(keys) != len(values):
+ raise Exception("Parallel keys and values lists must have the same length")
+ return dict(zip(keys, values))
+
+
+def dict_to_key_value_lists(d: Dict[Any, Any]) -> Tuple[List[Any], List[Any]]:
+ """
+ >>> d = {'name': 'scott', 'phone': '555-1212', 'address': '123 main st.', 'zip': '12345'}
+ >>> (k, v) = dict_to_key_value_lists(d)
+ >>> k
+ ['name', 'phone', 'address', 'zip']
+ >>> v
+ ['scott', '555-1212', '123 main st.', '12345']
+
+ """
+ r = ([], [])
+ for (k, v) in d.items():
+ r[0].append(k)
+ r[1].append(v)
+ return r
+
+
if __name__ == '__main__':
import doctest
doctest.testmod()