Upload utils.py with huggingface_hub

utils.py

@@ -1,7 +1,73 @@
+from typing import Dict, Any
+from queue import Queue
+
+
+def is_done(query):
+    return query is None or len(query) == 0 or query == '/'
+
+
+def is_wildcard(query):
+    return query == '*'
+
+
+def is_int(query):
+    return query.isdigit()
 
 class Singleton(type):
     _instances = {}
     def __call__(cls, *args, **kwargs):
         if cls not in cls._instances:
             cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
-        return cls._instances[cls]
+        return cls._instances[cls]
+
+
+def flatten_dict(d: Dict[str, Any], parent_key: str = "", sep: str = "_") -> Dict[str, Any]:
+    items = []
+    for k, v in d.items():
+        new_key = parent_key + sep + k if parent_key else k
+        if isinstance(v, dict):
+            items.extend(flatten_dict(v, new_key, sep=sep).items())
+        else:
+            items.append((new_key, v))
+    return dict(items)
+
+
+def dict_query(dic, query_path):
+    results = []
+    tasks = Queue()
+
+    tasks.put((dic, query_path, ''))
+
+    while not tasks.empty():
+
+        d, q, p = tasks.get()
+
+        if is_done(q):
+            results.append(d)
+        else:
+
+            parts = q.split('/', maxsplit=1)
+
+            current = parts[0]
+            nexts = parts[1] if len(parts) > 1 else None
+
+            if is_wildcard(current):
+                assert isinstance(d, list), f'cannot query wildcard "*" on non-list object: {d} (in path: {p})'
+                for item in d:
+                    tasks.put((item, nexts, p + f'/{item}'))
+
+            elif is_int(current):
+                assert isinstance(d, list), f'cannot query integer "{current}" on non-list object: {d} (in path: {p})'
+                tasks.put((d[int(current)], nexts, p + f'/{current}'))
+
+            elif not isinstance(d, dict):
+                raise ValueError(f'cannot query "{current}" on non-dict object: {d} (in path: {p})')
+            else:
+                if current not in d:
+                    raise ValueError(f'"{current}" not in dict object: {d} (in path: {p})')
+                tasks.put((d[current], nexts, p + f'/{current}'))
+
+    if len(results) == 0:
+        raise ValueError(f'query "{query_path}" did not match any item in dict: {dic}')
+
+    return results if len(results) > 1 else results[0]