2023-07-12 04:52:42 +01:00
|
|
|
import copy
|
2022-06-16 14:07:12 +01:00
|
|
|
import functools
|
2022-07-24 14:38:16 +01:00
|
|
|
from itertools import zip_longest
|
2022-07-09 11:06:48 +01:00
|
|
|
from typing import Iterator
|
2022-06-16 14:07:12 +01:00
|
|
|
|
|
|
|
|
|
|
|
def polling(func):
|
|
|
|
"""
|
|
|
|
Offers memoization for a set into get operation.
|
|
|
|
|
|
|
|
If sync clear dirty parameters before fetching new value.
|
|
|
|
|
|
|
|
Useful for loop getting if not running callbacks
|
|
|
|
"""
|
|
|
|
|
|
|
|
@functools.wraps(func)
|
|
|
|
def wrapper(*args, **kwargs):
|
|
|
|
get = func.__name__ == "get"
|
|
|
|
mb_get = func.__name__ == "get_buttonstatus"
|
|
|
|
remote, *remaining = args
|
|
|
|
|
|
|
|
if get:
|
|
|
|
param, *rem = remaining
|
|
|
|
elif mb_get:
|
|
|
|
id, mode, *rem = remaining
|
|
|
|
param = f"mb_{id}_{mode}"
|
|
|
|
|
|
|
|
if param in remote.cache:
|
|
|
|
return remote.cache.pop(param)
|
|
|
|
if remote.sync:
|
|
|
|
remote.clear_dirty()
|
|
|
|
return func(*args, **kwargs)
|
|
|
|
|
|
|
|
return wrapper
|
|
|
|
|
|
|
|
|
|
|
|
def script(func):
|
|
|
|
"""Convert dictionary to script"""
|
|
|
|
|
|
|
|
def wrapper(*args):
|
|
|
|
remote, script = args
|
|
|
|
if isinstance(script, dict):
|
|
|
|
params = ""
|
|
|
|
for key, val in script.items():
|
|
|
|
obj, m2, *rem = key.split("-")
|
|
|
|
index = int(m2) if m2.isnumeric() else int(*rem)
|
|
|
|
params += ";".join(
|
|
|
|
f"{obj}{f'.{m2}stream' if not m2.isnumeric() else ''}[{index}].{k}={int(v) if isinstance(v, bool) else v}"
|
|
|
|
for k, v in val.items()
|
|
|
|
)
|
|
|
|
params += ";"
|
|
|
|
script = params
|
|
|
|
return func(remote, script)
|
|
|
|
|
|
|
|
return wrapper
|
2022-07-09 11:06:48 +01:00
|
|
|
|
|
|
|
|
|
|
|
def comp(t0: tuple, t1: tuple) -> Iterator[bool]:
|
|
|
|
"""
|
|
|
|
Generator function, accepts two tuples.
|
|
|
|
|
|
|
|
Evaluates equality of each member in both tuples.
|
|
|
|
"""
|
|
|
|
for a, b in zip(t0, t1):
|
|
|
|
yield a == b
|
2022-07-24 14:38:16 +01:00
|
|
|
|
|
|
|
|
|
|
|
def grouper(n, iterable, fillvalue=None):
|
|
|
|
"""
|
|
|
|
Group elements of an iterable by sets of n length
|
|
|
|
"""
|
|
|
|
args = [iter(iterable)] * n
|
|
|
|
return zip_longest(fillvalue=fillvalue, *args)
|
2023-07-12 04:52:42 +01:00
|
|
|
|
|
|
|
|
|
|
|
def deep_merge(dict1, dict2):
|
|
|
|
"""Generator function for deep merging two dicts"""
|
|
|
|
for k in set(dict1) | set(dict2):
|
|
|
|
if k in dict1 and k in dict2:
|
|
|
|
if isinstance(dict1[k], dict) and isinstance(dict2[k], dict):
|
|
|
|
yield k, dict(deep_merge(dict1[k], dict2[k]))
|
|
|
|
else:
|
|
|
|
yield k, dict2[k]
|
|
|
|
elif k in dict1:
|
|
|
|
yield k, dict1[k]
|
|
|
|
else:
|
|
|
|
yield k, dict2[k]
|