Skip to content

Reference

param_scope(*args, **kwargs)

Bases: _HyperParameter

A thread-safe hyperparameter context scope

Examples:

create new param_scope

>>> ps = param_scope(a="a", b="b")           # create from call arguments
>>> ps = param_scope(**{"a": "a", "b": "b"}) # create from a dict

read parameters from param_scope

>>> ps.a() # read parameter
'a'
>>> ps.c("c")  # read parameter with default value if missing
'c'
>>> ps.c | "c" # another way for reading missing parameters
'c'

param_scope as a context scope

>>> with param_scope(**{"a": "a"}) as ps:
...     print(ps.a())
a

read parameter from param_scope in a function

>>> def foo():
...    with param_scope() as ps:
...        return ps.a()
>>> with param_scope(**{"a": "a", "b": "b"}) as ps:
...     foo() # foo should get param_scope using a with statement
'a'

modify parameters in nested scopes

>>> with param_scope.empty(**{'a': 1, 'b': 2}) as ps:
...     _repr_dict(ps.storage().storage())
...     with param_scope(**{'b': 3}) as ps:
...         _repr_dict(ps.storage().storage())
...     with param_scope() as ps:
...         _repr_dict(ps.storage().storage())
[('a', 1), ('b', 2)]
[('a', 1), ('b', 3)]
[('a', 1), ('b', 2)]

use object-style parameter key in param_scope

>>> with param_scope(**{"a.b.c": [1,2]}) as ps:
...     ps.a.b.c()
[1, 2]

access parameter with param_scope

>>> with param_scope(x=1):
...     param_scope.x(2) # read parameter
...     param_scope.y(2) # read a missing parameter with default value
...     param_scope.y | 2
...     param_scope.z = 3
...     param_scope.z | 0
1
2
2
3

convert param_scope to dict:

>>> ps = param_scope.empty(a=1, b=2)
>>> _repr_dict(dict(ps))
[('a', 1), ('b', 2)]
Source code in hyperparameter/api.py
def __init__(self, *args: str, **kwargs: Any) -> None:
    super().__init__()
    self.update(kwargs)
    for line in args:
        if "=" in line:
            k, v = line.split("=", 1)
            self.put(k, v)

__enter__()

enter a param_scope context

Examples:

>>> with param_scope():
...     param_scope.p = "origin"
...     with param_scope(**{"p": "origin"}) as ps:
...         ps.storage().storage()      # outer scope
...         with param_scope() as ps:   # unmodified scope
...             ps.storage().storage()  # inner scope
...         with param_scope(**{"p": "modified"}) as ps: # modified scope
...             ps.storage().storage()  # inner scope with modified params
...         _ = param_scope(**{"p": "modified"}) # not used in with ctx
...         with param_scope() as ps:   # unmodified scope
...             ps.storage().storage()  # inner scope
{'p': 'origin'}
{'p': 'origin'}
{'p': 'modified'}
{'p': 'origin'}
Source code in hyperparameter/api.py
def __enter__(self) -> "param_scope":
    """enter a `param_scope` context

    Examples
    --------
    >>> with param_scope():
    ...     param_scope.p = "origin"
    ...     with param_scope(**{"p": "origin"}) as ps:
    ...         ps.storage().storage()      # outer scope
    ...         with param_scope() as ps:   # unmodified scope
    ...             ps.storage().storage()  # inner scope
    ...         with param_scope(**{"p": "modified"}) as ps: # modified scope
    ...             ps.storage().storage()  # inner scope with modified params
    ...         _ = param_scope(**{"p": "modified"}) # not used in with ctx
    ...         with param_scope() as ps:   # unmodified scope
    ...             ps.storage().storage()  # inner scope
    {'p': 'origin'}
    {'p': 'origin'}
    {'p': 'modified'}
    {'p': 'origin'}
    """

    self._storage.enter()
    return self

current() staticmethod

get current param_scope

Examples:

>>> with param_scope(a=1) as ps:
...     param_scope.current().a("empty") # read `a` from current `param_scope`
'1'
>>> with param_scope() as ps1:
...     with param_scope(a=1) as ps2:
...         param_scope.current().a = 2  # set parameter `a` = 2
...         param_scope.a("empty")       # read `a` in `ps2`
...     param_scope.a("empty")           # read `a` in `ps1`, where `a` is not set
'2'
'empty'
Source code in hyperparameter/api.py
@staticmethod
def current() -> "param_scope":
    """get current `param_scope`

    Examples
    --------
    >>> with param_scope(a=1) as ps:
    ...     param_scope.current().a("empty") # read `a` from current `param_scope`
    '1'

    >>> with param_scope() as ps1:
    ...     with param_scope(a=1) as ps2:
    ...         param_scope.current().a = 2  # set parameter `a` = 2
    ...         param_scope.a("empty")       # read `a` in `ps2`
    ...     param_scope.a("empty")           # read `a` in `ps1`, where `a` is not set
    '2'
    'empty'
    """
    retval = param_scope()
    retval._storage = TLSKVStorage.current()
    return retval

empty(*args, **kwargs) staticmethod

create an empty param_scope.

Examples:

>>> with param_scope(a="not empty") as ps: # start a new param_scope `a` = 'not empty'
...     param_scope.a("empty")             # read parameter `a`
...     with param_scope.empty() as ps2:   # parameter `a` is cleared in ps2
...         param_scope.a("empty")         # read parameter `a` = 'empty'
'not empty'
'empty'
Source code in hyperparameter/api.py
@staticmethod
def empty(*args: str, **kwargs: Any) -> "param_scope":
    """create an empty `param_scope`.

    Examples
    --------
    >>> with param_scope(a="not empty") as ps: # start a new param_scope `a` = 'not empty'
    ...     param_scope.a("empty")             # read parameter `a`
    ...     with param_scope.empty() as ps2:   # parameter `a` is cleared in ps2
    ...         param_scope.a("empty")         # read parameter `a` = 'empty'
    'not empty'
    'empty'
    """
    retval = param_scope().clear().update(kwargs)
    for line in args:
        if "=" in line:
            k, v = line.split("=", 1)
            retval.put(k, v)
    return retval

init(params=None) staticmethod

init param_scope for a new thread.

Source code in hyperparameter/api.py
@staticmethod
def init(params: Optional[Dict[str, Any]] = None) -> None:
    """init param_scope for a new thread."""
    if params is None:
        params = {}
    param_scope(**params).__enter__()

auto_param(name_or_func)

auto_param(func: Callable) -> Callable
auto_param(name: str) -> Callable[[Callable], Callable]

Convert keyword arguments into hyperparameters

Examples:

>>> @auto_param
... def foo(a, b=2, c='c', d=None):
...     print(a, b, c, d)
>>> foo(1)
1 2 c None
>>> with param_scope('foo.b=3'):
...     foo(2)
2 3 c None

classes are also supported:

>>> @auto_param
... class foo:
...     def __init__(self, a, b=2, c='c', d=None):
...         print(a, b, c, d)
>>> obj = foo(1)
1 2 c None
>>> with param_scope('foo.b=3'):
...     obj = foo(2)
2 3 c None
>>> @auto_param('myns.foo.params')
... def foo(a, b=2, c='c', d=None):
...     print(a, b, c, d)
>>> foo(1)
1 2 c None
>>> with param_scope('myns.foo.params.b=3'):
...     foo(2)
2 3 c None
>>> with param_scope('myns.foo.params.b=3'):
...     param_scope.myns.foo.params.b = 4
...     foo(2)
2 4 c None
Source code in hyperparameter/api.py
def auto_param(
    name_or_func: Union[str, Callable, None],
) -> Union[Callable, Callable[[Callable], Callable]]:
    """Convert keyword arguments into hyperparameters

    Examples
    --------

    >>> @auto_param
    ... def foo(a, b=2, c='c', d=None):
    ...     print(a, b, c, d)

    >>> foo(1)
    1 2 c None

    >>> with param_scope('foo.b=3'):
    ...     foo(2)
    2 3 c None

    classes are also supported:
    >>> @auto_param
    ... class foo:
    ...     def __init__(self, a, b=2, c='c', d=None):
    ...         print(a, b, c, d)

    >>> obj = foo(1)
    1 2 c None

    >>> with param_scope('foo.b=3'):
    ...     obj = foo(2)
    2 3 c None

    >>> @auto_param('myns.foo.params')
    ... def foo(a, b=2, c='c', d=None):
    ...     print(a, b, c, d)
    >>> foo(1)
    1 2 c None

    >>> with param_scope('myns.foo.params.b=3'):
    ...     foo(2)
    2 3 c None

    >>> with param_scope('myns.foo.params.b=3'):
    ...     param_scope.myns.foo.params.b = 4
    ...     foo(2)
    2 4 c None
    """

    if callable(name_or_func):
        return auto_param(None)(name_or_func)

    if has_rust_backend:

        def hashed_wrapper(func: Callable) -> Callable:
            predef_kws: Dict[str, int] = {}
            predef_defaults: Dict[str, Any] = {}

            if name_or_func is None:
                namespace = func.__name__
            else:
                namespace = name_or_func

            signature = inspect.signature(func)
            for k, v in signature.parameters.items():
                if v.default != v.empty:
                    name = "{}.{}".format(namespace, k)
                    predef_kws[k] = xxh64(name)
                    predef_defaults[k] = v.default

            @functools.wraps(func)
            def inner(*arg: Any, **kws: Any) -> Any:
                with param_scope() as hp:
                    for k, v in predef_kws.items():
                        if k not in kws:
                            try:
                                val = hp._storage.get_entry(v)
                                kws[k] = _coerce_with_default(val, predef_defaults[k])
                            except ValueError:
                                pass
                    return func(*arg, **kws)

            inner._auto_param_namespace = namespace  # type: ignore[attr-defined]
            inner._auto_param_wrapped = func  # type: ignore[attr-defined]
            return inner

        return hashed_wrapper

    def wrapper(func: Callable) -> Callable:
        predef_kws: Dict[str, str] = {}
        predef_val: Dict[str, Any] = {}

        if name_or_func is None:
            namespace = func.__name__
        else:
            namespace = name_or_func

        signature = inspect.signature(func)
        for k, v in signature.parameters.items():
            if v.default != v.empty:
                name = "{}.{}".format(namespace, k)
                predef_kws[k] = name
                predef_val[name] = v.default

        @functools.wraps(func)
        def inner(*arg: Any, **kws: Any) -> Any:
            with param_scope() as hp:
                local_params: Dict[str, Any] = {}
                for k, v in predef_kws.items():
                    if k not in kws:
                        val = getattr(hp(), v).get_or_else(predef_val[v])
                        kws[k] = val
                        local_params[v] = hp.get(v)
                    else:
                        local_params[v] = predef_val[v]
                return func(*arg, **kws)

        inner._auto_param_namespace = namespace  # type: ignore[attr-defined]
        inner._auto_param_wrapped = func  # type: ignore[attr-defined]
        return inner

    return wrapper

launch(func=None, *, _caller_globals=None, _caller_locals=None)

Launch CLI for @auto_param functions.

  • launch(f): expose a single @auto_param function f as CLI.
  • launch(): expose all @auto_param functions in the caller module as subcommands.
Source code in hyperparameter/api.py
def launch(func: Optional[Callable] = None, *, _caller_globals=None, _caller_locals=None) -> None:
    """Launch CLI for @auto_param functions.

    - launch(f): expose a single @auto_param function f as CLI.
    - launch(): expose all @auto_param functions in the caller module as subcommands.
    """
    if _caller_globals is None or _caller_locals is None:
        caller_frame = inspect.currentframe().f_back  # type: ignore
        caller_globals = caller_frame.f_globals if caller_frame else {}
        caller_locals = caller_frame.f_locals if caller_frame else {}
    else:
        caller_globals = _caller_globals
        caller_locals = _caller_locals

    if func is None:
        seen_ids = set()
        candidates = []
        for obj in list(caller_locals.values()) + list(caller_globals.values()):
            if not callable(obj):
                continue
            ns = getattr(obj, "_auto_param_namespace", None)
            if not isinstance(ns, str):
                continue
            # Skip private helpers (e.g., _foo) when exposing subcommands.
            name = getattr(obj, "__name__", "")
            if isinstance(name, str) and name.startswith("_"):
                continue
            oid = id(obj)
            if oid in seen_ids:
                continue
            seen_ids.add(oid)
            candidates.append(obj)
        if not candidates:
            raise RuntimeError("No @auto_param functions found to launch.")

        if len(candidates) == 1:
            import sys

            func = candidates[0]
            parser = _build_parser_for_func(func)
            argv = sys.argv[1:]
            if argv and argv[0] == func.__name__:
                argv = argv[1:]
            args = parser.parse_args(argv)
            args_dict = vars(args)
            defines = args_dict.pop("define", [])
            with param_scope(*defines):
                return func(**args_dict)

        parser = argparse.ArgumentParser(description="hyperparameter auto-param CLI")
        subparsers = parser.add_subparsers(dest="command", required=True)
        func_map: Dict[str, Callable] = {}
        for f in candidates:
            sub = subparsers.add_parser(f.__name__, help=f.__doc__)
            func_map[f.__name__] = f
            sub.add_argument("-D", "--define", nargs="*", default=[], action="extend", help="Override params, e.g., a.b=1")
            sig = inspect.signature(f)
            param_help = _parse_param_help(f.__doc__)
            for name, param in sig.parameters.items():
                if param.default is inspect._empty:
                    sub.add_argument(name, type=param.annotation if param.annotation is not inspect._empty else str, help=param_help.get(name))
                else:
                    arg_type = _arg_type_from_default(param.default)
                    help_text = param_help.get(name)
                    if help_text:
                        help_text = f"{help_text} (default: {param.default})"
                    else:
                        help_text = f"(default from auto_param: {param.default})"
                    sub.add_argument(
                        f"--{name}",
                        dest=name,
                        type=arg_type,
                        default=argparse.SUPPRESS,
                        help=help_text,
                    )
        args = parser.parse_args()
        args_dict = vars(args)
        cmd = args_dict.pop("command")
        defines = args_dict.pop("define", [])
        target = func_map[cmd]
        with param_scope(*defines):
            # Freeze first so new threads spawned inside target inherit these overrides.
            param_scope.frozen()
            return target(**args_dict)

    if not hasattr(func, "_auto_param_namespace"):
        raise ValueError("launch() expects a function decorated with @auto_param")
    parser = _build_parser_for_func(func)
    args = parser.parse_args()
    args_dict = vars(args)
    defines = args_dict.pop("define", [])
    with param_scope(*defines):
        param_scope.frozen()
        return func(**args_dict)

run_cli(func=None)

Alias for launch() with a less collision-prone name.

Source code in hyperparameter/api.py
def run_cli(func: Optional[Callable] = None) -> None:
    """Alias for launch() with a less collision-prone name."""
    caller_frame = inspect.currentframe().f_back  # type: ignore
    caller_globals = caller_frame.f_globals if caller_frame else {}
    caller_locals = caller_frame.f_locals if caller_frame else {}
    return launch(func, _caller_globals=caller_globals, _caller_locals=caller_locals)