core: feature: guess module stats from typing annotations

my/core/__main__.py

@@ -240,6 +240,7 @@ def modules_check(args: Namespace) -> None:
     tabulate_warnings()
 
     from .util import get_stats, HPIModule
+    from .stats import guess_stats
 
     mods: Iterable[HPIModule]
     if module is None:
@@ -265,7 +266,12 @@ def modules_check(args: Namespace) -> None:
             continue
 
         info(f'{color.GREEN}OK{color.RESET}  : {m:<50}')
+        # first try explicitly defined stats function:
         stats = get_stats(m)
+        if stats is None:
+            # then try guessing.. not sure if should log somehow?
+            stats = guess_stats(m)
+
         if stats is None:
             eprint("       - no 'stats' function, can't check the data")
             # todo point to a readme on the module structure or something?

my/core/stats.py

@@ -0,0 +1,97 @@
+'''
+Helpers for hpi doctor/stats functionality.
+'''
+import collections
+import importlib
+import inspect
+import sys
+import typing
+from typing import Optional
+
+from .common import StatsFun, Stats, stat
+
+
+# TODO maybe could be enough to annotate OUTPUTS or something like that?
+# then stats could just use them as hints?
+def guess_stats(module_name: str) -> Optional[StatsFun]:
+    module = importlib.import_module(module_name)
+    mfunctions = inspect.getmembers(module, inspect.isfunction)
+    functions = {k: v for k, v in mfunctions if is_data_provider(v)}
+    if len(functions) == 0:
+        return None
+    def auto_stats() -> Stats:
+        return {k: stat(v) for k, v in functions.items()}
+    return auto_stats
+
+
+def is_data_provider(fun) -> bool:
+    """
+    1. returns iterable or something like that
+    2. takes no arguments? (otherwise not callable by stats anyway?)
+    """
+    if fun is None:
+        return False
+    # todo. uh.. very similar to what cachew is trying to do?
+    try:
+        sig = inspect.signature(fun)
+    except ValueError:  # not a function?
+        return False
+
+    if len(sig.parameters) > 0:
+        return False
+    return_type = sig.return_annotation
+    return type_is_iterable(return_type)
+
+
+def test_is_data_provider() -> None:
+    idp = is_data_provider
+    assert not idp(None)
+    assert not idp(int)
+
+    def no_return_type():
+        return [1, 2 ,3]
+    assert not idp(no_return_type)
+
+    lam = lambda: [1, 2]
+    assert not idp(lam)
+
+    def has_extra_args(count) -> typing.List[int]:
+        return list(range(count))
+    assert not idp(has_extra_args)
+
+    def has_return_type() -> typing.Sequence[str]:
+        return ['a', 'b', 'c']
+    assert idp(has_return_type)
+
+
+def type_is_iterable(type_spec) -> bool:
+    if sys.version_info[1] < 8:
+        # there is no get_origin before 3.8, and retrofitting gonna be a lot of pain
+        return any(x in str(type_spec) for x in ['List', 'Sequence', 'Iterable', 'Iterator'])
+    origin = typing.get_origin(type_spec)
+    if origin is None:
+        return False
+
+    # explicitly exclude dicts... not sure?
+    if issubclass(origin, collections.abc.Mapping):
+        return False
+
+    if issubclass(origin, collections.abc.Iterable):
+        return True
+
+    return False
+
+
+# todo docstring test?
+def test_type_is_iterable() -> None:
+    from typing import List, Sequence, Iterable, Dict, Any
+
+    fun = type_is_iterable
+    assert not fun(None)
+    assert not fun(int)
+    assert not fun(Any)
+    assert not fun(Dict[int, int])
+
+    assert fun(List[int])
+    assert fun(Sequence[Dict[str, str]])
+    assert fun(Iterable[Any])

tests/core.py

@@ -17,3 +17,4 @@ from my.core.error          import *
 from my.core.util           import *
 from my.core.discovery_pure import *
 from my.core.types          import *
+from my.core.stats          import *