Skip to content

Reference

This part of the project documentation focuses on an information-oriented approach. Use it as a reference for the technical implementation of the async_btree project code.

Declare async btree api.

AsyncInnerFunction = Callable[[], Awaitable[Any]] module-attribute

Function signature of async function implementation.

CallableFunction = Union[Callable[Ellipsis, Awaitable[Any]], Callable] module-attribute

Something callable with or without async.

FAILURE = not SUCCESS module-attribute

Failure constant.

SUCCESS = True module-attribute

Success constant.

BTreeRunner

A context manager that call multiple async btree function in same context from sync framework.

asyncio provide a Runner (python >= 3.11) to call several top-level async functions in the SAME context.

The goal here is to hide underlaying asyncio framework.

This function cannot be called when another asyncio event loop is running in the same thread.

This function always creates a new event loop or Kernel and closes it at the end. It should be used as a main entry point for asyncio programs, and should ideally only be called once.

Source code in async_btree/runner.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class BTreeRunner:
    """A context manager that call multiple async btree function in same context from sync framework.

    `asyncio` provide a Runner (python >= 3.11) to call several top-level async functions in the SAME context.

    The goal here is to hide underlaying asyncio framework.

    This function cannot be called when another asyncio event loop is running in the same thread.

    This function always creates a new event loop or Kernel and closes it at the end.
    It should be used as a main entry point for asyncio programs, and should ideally only be called once.

    """

    def __init__(self, disable_curio: bool = False) -> None:
        """Create a runner to call ultiple async btree function in same context from existing sync framework.

        Args:
            disable_curio (bool, optional): Force usage of `asyncio` Defaults to False.

        Raises:
            RuntimeError: if python version is below 3.11 and disable_curio is set.
        """
        self._has_curio = has_curio() and not disable_curio
        self._context: Optional[Context] = None
        # curio support
        self._kernel: Optional[ContextManager] = None
        # asyncio support
        if not self._has_curio and sys.version_info.minor < 11:
            raise RuntimeError("asyncio support only for python 3.11")
        self._runner = None

    def __enter__(self):
        self._context = copy_context()

        if self._has_curio:
            from curio import Kernel

            self._kernel = Kernel()
        else:
            from asyncio import Runner

            self._kernel = Runner()

        self._kernel.__enter__()  # type: ignore

        return self

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            self._kernel.__exit__(exc_type, exc_value, traceback)  # type: ignore
        finally:
            self._kernel = None
            self._context = None

    def run(self, target: Callable[..., Awaitable[R]], *args, **kwargs) -> R:
        """Run an async btree coroutine in a same context.

        Args:
            target (Callable[..., Awaitable[R]]): coroutine

        Raises:
            RuntimeError: if context is not initialized

        Returns:
            R: result
        """
        if not self._kernel:
            raise RuntimeError("run method must be invoked inside a context.")
        coro = target(*args, **kwargs)
        if self._has_curio:
            return self._context.run(self._kernel.run, coro)  # type: ignore
        return self._kernel.run(coro, context=self._context)  # type: ignore

__init__(disable_curio=False)

Create a runner to call ultiple async btree function in same context from existing sync framework.

Parameters:

Name Type Description Default
disable_curio bool

Force usage of asyncio Defaults to False.

 False

Raises:

Type Description
RuntimeError

if python version is below 3.11 and disable_curio is set.
Source code in async_btree/runner.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(self, disable_curio: bool = False) -> None:
    """Create a runner to call ultiple async btree function in same context from existing sync framework.

    Args:
        disable_curio (bool, optional): Force usage of `asyncio` Defaults to False.

    Raises:
        RuntimeError: if python version is below 3.11 and disable_curio is set.
    """
    self._has_curio = has_curio() and not disable_curio
    self._context: Optional[Context] = None
    # curio support
    self._kernel: Optional[ContextManager] = None
    # asyncio support
    if not self._has_curio and sys.version_info.minor < 11:
        raise RuntimeError("asyncio support only for python 3.11")
    self._runner = None

run(target, *args, **kwargs)

Run an async btree coroutine in a same context.

Parameters:

Name Type Description Default
target Callable[..., Awaitable[R]]

coroutine

 required

Raises:

Type Description
RuntimeError

if context is not initialized

Returns:

Name Type Description
R R

result
Source code in async_btree/runner.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def run(self, target: Callable[..., Awaitable[R]], *args, **kwargs) -> R:
    """Run an async btree coroutine in a same context.

    Args:
        target (Callable[..., Awaitable[R]]): coroutine

    Raises:
        RuntimeError: if context is not initialized

    Returns:
        R: result
    """
    if not self._kernel:
        raise RuntimeError("run method must be invoked inside a context.")
    coro = target(*args, **kwargs)
    if self._has_curio:
        return self._context.run(self._kernel.run, coro)  # type: ignore
    return self._kernel.run(coro, context=self._context)  # type: ignore

ControlFlowException

Bases: Exception

ControlFlowException exception is a decorator on a real exception.

This will ensure that assert ControlFlowException.__bool__ == False. This permit to return exception as a 'FAILURE' status.

Source code in async_btree/definition.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class ControlFlowException(Exception):
    """ControlFlowException exception is a decorator on a real exception.

    This will ensure that ```assert ControlFlowException.__bool__ == False```.
    This permit to return exception as a 'FAILURE' status.
    """

    def __init__(self, exception: Exception):
        super().__init__()

        self.exception = exception

    def __bool__(self):
        return False

    def __repr__(self):
        return self.exception.__repr__()

    def __str__(self):
        return self.exception.__str__()

    @classmethod
    def instanciate(cls, exception: Exception):
        # this methods simplify usage of hierarchical call tree.
        return exception if isinstance(exception, ControlFlowException) else ControlFlowException(exception=exception)

Node

Bases: NamedTuple

Node aggregate node definition implemented with NamedTuple.

A Node is used to keep information on name, properties, and relations ship between a hierachical construct of functions. It's like an instance of NodeMetadata.

Attributes:

Name Type Description
name str

named operation.
properties List[Tuple[str, Any]]

a list of tuple (name, value) for definition.
edges List[Tuple[str, List[Any]]]

a list of tuple (name, node list) for definition.
Notes

Edges attribut should be edges: List[Tuple[str, List['Node']]] But it is impossible for now, see mypy issues 731

Source code in async_btree/analyze.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class Node(NamedTuple):
    """Node aggregate node definition implemented with NamedTuple.

    A Node is used to keep information on name, properties, and relations ship
    between a hierachical construct of functions.
    It's like an instance of NodeMetadata.

    Attributes:
        name (str): named operation.
        properties (List[Tuple[str, Any]]): a list of tuple (name, value) for definition.
        edges (List[Tuple[str, List[Any]]]): a list of tuple (name, node list) for
            definition.

    Notes:
        Edges attribut should be edges: ```List[Tuple[str, List['Node']]]```
        But it is impossible for now, see [mypy issues 731](https://github.com/python/mypy/issues/731)
    """

    name: str
    properties: List[Tuple[str, Any]]
    # edges: List[Tuple[str, List['Node']]]
    # https://github.com/python/mypy/issues/731
    edges: List[Tuple[str, List[Any]]]

    def __str__(self):
        return stringify_analyze(target=self)

NodeMetadata

Bases: NamedTuple

NodeMetadata is our node definition.

A NodeMetadata is used to keep information on name, properties name, and relations ship name between a hierachical construct of functions.

This permit us to print or analyze all information of a behaviour tree.

Attributes:

Name Type Description
name str

named operation
properties List[str]

a list of property name (an int value, ...).
edges List[str]

a list of member name which act as edges (a child, ...).
Source code in async_btree/definition.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class NodeMetadata(NamedTuple):
    """NodeMetadata is our node definition.

    A NodeMetadata is used to keep information on name, properties name,
    and relations ship name between a hierachical construct of functions.

    This permit us to print or analyze all information of a behaviour tree.

    Attributes:
        name (str): named operation
        properties (List[str]): a list of property name (an int value, ...).
        edges (List[str]): a list of member name which act as edges (a child, ...).

    """

    name: str
    properties: Optional[List[str]] = None
    edges: Optional[List[str]] = None

    @classmethod
    def alias(cls, name: str, node: NodeMetadata, properties: Optional[List[str]] = None) -> NodeMetadata:
        return NodeMetadata(name=name, properties=properties if properties else node.properties, edges=node.edges)

action(target, **kwargs)

Declare an action leaf.

Action is an awaitable closure of specified function, (See alias function).

Parameters:

Name Type Description Default
target CallableFunction

awaitable function

 required
kwargs

optional kwargs argument to pass on target function

 {}

Returns:

Type Description
AsyncInnerFunction

an awaitable function.

Raises:

Type Description
ControlFlowException

if error occurs
Source code in async_btree/leaf.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def action(target: CallableFunction, **kwargs) -> AsyncInnerFunction:
    """Declare an action leaf.

    Action is an awaitable closure of specified function,
    (See alias function).

    Args:
        target (CallableFunction): awaitable function
        kwargs: optional kwargs argument to pass on target function

    Returns:
        (AsyncInnerFunction): an awaitable function.

    Raises:
        ControlFlowException : if error occurs

    """

    _target = to_async(target)

    @node_metadata(properties=['_target'])
    async def _action():
        try:
            return await _target(**kwargs)
        except Exception as e:
            raise ControlFlowException.instanciate(e)

    return _action

afilter(corofunc, iterable) async

Filter an iterable or an async iterable with an async function.

This simplify writing of filtering by a function on something iterable between 'async for ...' and 'for...' .

Parameters:

Name Type Description Default
corofunc Callable[[Any], Awaitable[bool]]

filter async function

 required
iterable Union[AsyncIterable, Iterable]

iterable or async iterable collection which will be applied.

 required

Returns:

Type Description
AsyncGenerator[T]

an async iterator of item which satisfy corofunc(item) == True
Example

[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]

Source code in async_btree/utils.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def afilter(
    corofunc: Callable[[Any], Awaitable[bool]], iterable: Union[AsyncIterable, Iterable]
) -> AsyncGenerator[T, None]:
    """Filter an iterable or an async iterable with an async function.

    This simplify writing of filtering by a function on something iterable
    between 'async for ...' and 'for...' .

    Args:
        corofunc (Callable[[Any], Awaitable[bool]]): filter async function
        iterable (Union[AsyncIterable, Iterable]): iterable or async iterable collection
            which will be applied.

    Returns:
        (AsyncGenerator[T]): an async iterator of item which satisfy corofunc(item) == True

    Example:
        ```[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]```

    """
    if isinstance(iterable, AsyncIterable):
        async for item in iterable:
            if await corofunc(item):
                yield item
    else:
        for item in iterable:
            if await corofunc(item):
                yield item

alias(child, name)

Define an alias on our child.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required
name str

name of function tree

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/decorator.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def alias(child: CallableFunction, name: str) -> AsyncInnerFunction:
    """Define an alias on our child.

    Args:
        child (CallableFunction): child function to decorate
        name (str): name of function tree

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """

    _child = to_async(child)

    # we use a dedicted function to 'duplicate' the child reference
    @node_metadata(name=name)
    async def _alias():
        return await _child()

    return _alias

always_failure(child)

Produce a function which always return FAILURE value.

Note

If you wanna git a failure even if an exception occurs, you have to decorate child with ignore_exception, like this:

always_failure(child=ignore_exception(myfunction))

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return child result if is falsy else FAILURE.

Raises:

Type Description
ControlFlowException

if error occurs
Source code in async_btree/decorator.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def always_failure(child: CallableFunction) -> AsyncInnerFunction:  # -> Awaitable:
    """Produce a function which always return FAILURE value.

    Note:
        If you wanna git a failure even if an exception occurs, you have
        to decorate child with ignore_exception, like this:

        `always_failure(child=ignore_exception(myfunction))`

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return child result if is falsy
            else FAILURE.

    Raises:
        ControlFlowException : if error occurs

    """

    _child = to_async(child)

    @node_metadata()
    async def _always_failure():
        result: Any = FAILURE

        try:
            child_result = await _child()
            if not bool(child_result):
                result = child_result

        except Exception as e:
            raise ControlFlowException.instanciate(e)

        return result

    return _always_failure

always_success(child)

Create a node which always return SUCCESS value.

Note

If you wanna git a success even if an exception occurs, you have to decorate child with ignore_exception, like this:

always_success(child=ignore_exception(myfunction))

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return child result if it is truthy else SUCCESS.

Raises:

Type Description
ControlFlowException

if error occurs
Source code in async_btree/decorator.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def always_success(child: CallableFunction) -> AsyncInnerFunction:
    """Create a node which always return SUCCESS value.

    Note:
        If you wanna git a success even if an exception occurs, you have
        to decorate child with ignore_exception, like this:

        `always_success(child=ignore_exception(myfunction))`


    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return child result if it is truthy
            else SUCCESS.

    Raises:
        ControlFlowException : if error occurs

    """

    _child = to_async(child)

    @node_metadata()
    async def _always_success():
        result: Any = SUCCESS

        try:
            child_result = await _child()
            if bool(child_result):
                result = child_result

        except Exception as e:
            raise ControlFlowException.instanciate(e)

        return result

    return _always_success

amap(corofunc, iterable) async

Map an async function onto an iterable or an async iterable.

This simplify writing of mapping a function on something iterable between 'async for ...' and 'for...' .

Parameters:

Name Type Description Default
corofunc Callable[[Any], Awaitable[T]]

coroutine function

 required
iterable Union[AsyncIterable, Iterable]

iterable or async iterable collection which will be applied.

 required

Returns:

Type Description
AsyncGenerator[T, None]

AsyncGenerator[T]: an async iterator of corofunc(item)
Example

[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]

Source code in async_btree/utils.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
async def amap(
    corofunc: Callable[[Any], Awaitable[T]], iterable: Union[AsyncIterable, Iterable]
) -> AsyncGenerator[T, None]:
    """Map an async function onto an iterable or an async iterable.

    This simplify writing of mapping a function on something iterable
    between 'async for ...' and 'for...' .

    Args:
        corofunc (Callable[[Any], Awaitable[T]]): coroutine function
        iterable (Union[AsyncIterable, Iterable]): iterable or async iterable collection
            which will be applied.

    Returns:
        AsyncGenerator[T]: an async iterator of corofunc(item)

    Example:
        ```[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]```

    """
    if isinstance(iterable, AsyncIterable):
        async for item in iterable:
            yield await corofunc(item)
    else:
        for item in iterable:
            yield await corofunc(item)

condition(target, **kwargs)

Declare a condition leaf.

Condition is an awaitable closure of specified function.

Parameters:

Name Type Description Default
target CallableFunction

awaitable function which be evaluated as True/False.

 required
kwargs

optional kwargs argument to pass on target function

 {}

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/leaf.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def condition(target: CallableFunction, **kwargs) -> AsyncInnerFunction:
    """Declare a condition leaf.

    Condition is an awaitable closure of specified function.

    Args:
        target (CallableFunction):  awaitable function which be evaluated as True/False.
        kwargs: optional kwargs argument to pass on target function

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """
    return alias_node_metadata(
        name='condition', target=is_success(action(target=target, **kwargs)), properties=['target']
    )

decision(condition, success_tree, failure_tree=None)

Create a decision node.

If condition is meet, return evaluation of success_tree. Otherwise, it return SUCCESS or evaluation of failure_tree if setted.

Parameters:

Name Type Description Default
condition CallableFunction

awaitable condition

 required
success_tree CallableFunction

awaitable success tree which be evaluated if cond is Truthy

 required
failure_tree CallableFunction

awaitable failure tree which be evaluated if cond is Falsy (None per default)

 None

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def decision(
    condition: CallableFunction, success_tree: CallableFunction, failure_tree: Optional[CallableFunction] = None
) -> AsyncInnerFunction:
    """Create a decision node.

    If condition is meet, return evaluation of success_tree.
    Otherwise, it return SUCCESS or evaluation of failure_tree if setted.

    Args:
        condition (CallableFunction): awaitable condition
        success_tree (CallableFunction): awaitable success tree which be
            evaluated if cond is Truthy
        failure_tree (CallableFunction): awaitable failure tree which be
            evaluated if cond is Falsy (None per default)

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """

    _condition = to_async(condition)
    _success_tree = to_async(success_tree)
    _failure_tree = to_async(failure_tree) if failure_tree else None

    @node_metadata(edges=['_condition', '_success_tree', '_failure_tree'])
    async def _decision():
        if bool(await _condition()):
            return await _success_tree()
        if _failure_tree:
            return await _failure_tree()
        return SUCCESS

    return _decision

decorate(child, decorator, **kwargs)

Create a decorator.

Post process a child with specified decorator function. First argument of decorator function must be a child.

This method implement a simple lazy evaluation.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required
decorator CallableFunction

awaitable target decorator with profile 'decorator(child_result, **kwargs)'

 required
kwargs

optional keyed argument to pass to decorator function

 {}

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return decorator evaluation against child.
Source code in async_btree/decorator.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def decorate(child: CallableFunction, decorator: CallableFunction, **kwargs) -> AsyncInnerFunction:
    """Create a decorator.

    Post process a child with specified decorator function.
    First argument of decorator function must be a child.

    This method implement a simple lazy evaluation.

    Args:
        child (CallableFunction): child function to decorate
        decorator (CallableFunction): awaitable target decorator with profile 'decorator(child_result, **kwargs)'
        kwargs: optional keyed argument to pass to decorator function

    Returns:
      (AsyncInnerFunction): an awaitable function which
            return decorator evaluation against child.
    """

    _child = to_async(child)
    _decorator = to_async(decorator)

    @node_metadata(properties=['_decorator'])
    async def _decorate():
        return await _decorator(await _child(), **kwargs)

    return _decorate

fallback(children)

Execute tasks in sequence and succeed if one succeed or failed if all failed.

Often named 'selector', children can be seen as an ordered list starting from higthest priority to lowest priority.

Parameters:

Name Type Description Default
children List[CallableFunction]

list of Awaitable

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
def fallback(children: List[CallableFunction]) -> AsyncInnerFunction:
    """Execute tasks in sequence and succeed if one succeed or failed if all failed.

    Often named 'selector', children can be seen as an ordered list
        starting from higthest priority to lowest priority.

    Args:
        children (List[CallableFunction]): list of Awaitable

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """
    return alias_node_metadata(name='fallback', target=sequence(children, succes_threshold=min(1, len(children))))

ignore_exception(child)

Create a node which ignore runtime exception.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return child result
AsyncInnerFunction

or any exception with a falsy meaning in a ControlFlowException.
Source code in async_btree/decorator.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def ignore_exception(child: CallableFunction) -> AsyncInnerFunction:
    """Create a node which ignore runtime exception.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return child result
        or any exception with a falsy meaning in a ControlFlowException.

    """

    _child = to_async(child)

    @node_metadata()
    async def _ignore_exception():

        try:
            return await _child()

        except Exception as e:
            return ControlFlowException.instanciate(e)

    return _ignore_exception

inverter(child)

Invert node status.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return SUCCESS if child return FAILURE else SUCCESS
Source code in async_btree/decorator.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def inverter(child: CallableFunction) -> AsyncInnerFunction:
    """Invert node status.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return SUCCESS if child
            return FAILURE else SUCCESS
    """

    _child = to_async(child)

    @node_metadata()
    async def _inverter():
        return not bool(await _child())

    return _inverter

is_failure(child)

Create a conditional node which test if child fail.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return SUCCESS if child return FAILURE else FAILURE.
Source code in async_btree/decorator.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def is_failure(child: CallableFunction) -> AsyncInnerFunction:
    """Create a conditional node which test if child fail.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return SUCCESS if child
            return FAILURE else FAILURE.
    """

    _child = to_async(child)

    @node_metadata()
    async def _is_failure():
        return SUCCESS if not bool(await _child()) else FAILURE

    return _is_failure

is_success(child)

Create a conditional node which test if child success.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return SUCCESS if child return SUCCESS else FAILURE.
Source code in async_btree/decorator.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def is_success(child: CallableFunction) -> AsyncInnerFunction:
    """Create a conditional node which test if child success.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return SUCCESS if child
            return SUCCESS else FAILURE.
    """

    _child = to_async(child)

    @node_metadata()
    async def _is_success():
        return SUCCESS if bool(await _child()) else FAILURE

    return _is_success

node_metadata(name=None, properties=None, edges=None)

'node_metadata' is a function decorator which add meta information about node.

We add a property on decorated function named '__node_metadata'.

Parameters:

Name Type Description Default
name Optional[str]

override name of decorated function, default is function name left striped with '_'

 None
properties Optional[List[str]]

a list of property name ([] as default)

 None
edges Optional[List[str]]

a list of edges name (["child", "children"] as default)

 None

Returns:

Type Description
Callable[[Callable[P, R]], FunctionWithMetadata[P, R]]

the decorator function
Source code in async_btree/definition.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def node_metadata(
    name: Optional[str] = None, properties: Optional[List[str]] = None, edges: Optional[List[str]] = None
) -> Callable[[Callable[P, R]], FunctionWithMetadata[P, R]]:
    """'node_metadata' is a function decorator which add meta information about node.

    We add a property on decorated function named '__node_metadata'.

    Args:
        name (Optional[str]): override name of decorated function,
            default is function name left striped with '_'
        properties (Optional[List[str]]): a list of property name ([] as default)
        edges (Optional[List[str]]): a list of edges name
            (["child", "children"] as default)

    Returns:
        the decorator function

    """

    def decorate_function(function: Callable[P, R]) -> FunctionWithMetadata[P, R]:
        dfunc = _attr_decorator(function)

        dfunc.__node_metadata = getattr(
            dfunc,
            "__node_metadata",
            NodeMetadata(name=name if name else get_function_name(target=dfunc), properties=properties, edges=edges),
        )
        return cast(FunctionWithMetadata[P, R], dfunc)

    return decorate_function

repeat_until(condition, child)

Repeat child evaluation until condition is truthy.

Return last child evaluation or FAILURE if no evaluation occurs.

Parameters:

Name Type Description Default
condition CallableFunction

awaitable condition

 required
child CallableFunction

awaitable child

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def repeat_until(condition: CallableFunction, child: CallableFunction) -> AsyncInnerFunction:
    """Repeat child evaluation until condition is truthy.

    Return last child evaluation or FAILURE if no evaluation occurs.

    Args:
        condition (CallableFunction): awaitable condition
        child (CallableFunction): awaitable child

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """

    _child = to_async(child)
    _condition = to_async(condition)

    @node_metadata(edges=['_condition', '_child'])
    async def _repeat_until():
        result: Any = FAILURE
        while bool(await _condition()):
            result = await _child()

        return result

    return _repeat_until

retry(child, max_retry=3)

Retry child evaluation at most max_retry time on failure until child succeed.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required
max_retry int

max retry count (default 3), -1 mean infinite retry

 3

Returns:

Type Description
AsyncInnerFunction

an awaitable function which retry child evaluation at most max_retry time on failure until child succeed. If max_retry is reached, returns FAILURE or last exception.
Source code in async_btree/decorator.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def retry(child: CallableFunction, max_retry: int = 3) -> AsyncInnerFunction:
    """Retry child evaluation at most max_retry time on failure until child succeed.

    Args:
        child (CallableFunction): child function to decorate
        max_retry (int): max retry count (default 3), -1 mean infinite retry

    Returns:
        (AsyncInnerFunction): an awaitable function which retry child evaluation
            at most max_retry time on failure until child succeed.
            If max_retry is reached, returns FAILURE or last exception.
    """
    if not (max_retry > 0 or max_retry == -1):
        raise AssertionError('max_retry')

    _child = to_async(child)

    @node_metadata(properties=['max_retry'])
    async def _retry():
        retry_count = max_retry
        result: Any = FAILURE

        while not bool(result) and retry_count != 0:
            result = await _child()
            print(f"result : {result}")
            retry_count -= 1

        return result

    return _retry

retry_until_failed(child)

Retry child until failed.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which try to evaluate child until it failed.
Source code in async_btree/decorator.py 
291
292
293
294
295
296
297
298
299
300
301
302
def retry_until_failed(child: CallableFunction) -> AsyncInnerFunction:
    """Retry child until failed.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which try to evaluate child
            until it failed.
    """

    return alias_node_metadata(name='retry_until_failed', target=retry(child=inverter(child), max_retry=-1))

retry_until_success(child)

Retry child until success.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which try to evaluate child until it succeed.
Source code in async_btree/decorator.py 
278
279
280
281
282
283
284
285
286
287
288
def retry_until_success(child: CallableFunction) -> AsyncInnerFunction:
    """Retry child until success.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which try to evaluate child
            until it succeed.
    """
    return alias_node_metadata(name='retry_until_success', target=retry(child=child, max_retry=-1))

run(kernel, target, *args)

Curio run with independent contextvars.

This mimic asyncio framework behaviour. We use a contextvars per run rather than use one per task with from curio.task.ContextTask

copy_context().run(kernel.run, target, *args)
Source code in async_btree/utils.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
def run(kernel, target, *args):
    """Curio run with independent contextvars.

    This mimic asyncio framework behaviour.
    We use a contextvars per run rather than use one per task with `from curio.task.ContextTask`

    ```
    copy_context().run(kernel.run, target, *args)
    ```

    """
    warn('This method is deprecated.', DeprecationWarning, stacklevel=2)
    return copy_context().run(kernel.run, target, *args)

selector(children)

Synonym of fallback.

Source code in async_btree/control.py 
83
84
85
def selector(children: List[CallableFunction]) -> AsyncInnerFunction:
    """Synonym of fallback."""
    return alias_node_metadata(name='selector', target=sequence(children, succes_threshold=min(1, len(children))))

sequence(children, succes_threshold=None)

Return a function which execute children in sequence.

succes_threshold parameter generalize traditional sequence/fallback and must be in [0, len(children)]. Default value is (-1) means len(children)

if #success = succes_threshold, return a success

if #failure = len(children) - succes_threshold, return a failure

What we can return as value and keep sematic Failure/Success: - an array of previous result when success - last failure when fail

Parameters:

Name Type Description Default
children List[CallableFunction]

list of Awaitable

 required
succes_threshold int

succes threshold value

 None

Returns:

Type Description
AsyncInnerFunction

an awaitable function.

Raises:

Type Description
AssertionError

if succes_threshold is invalid
Source code in async_btree/control.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def sequence(children: List[CallableFunction], succes_threshold: Optional[int] = None) -> AsyncInnerFunction:
    """Return a function which execute children in sequence.

    succes_threshold parameter generalize traditional sequence/fallback and
    must be in [0, len(children)]. Default value is (-1) means len(children)

    if #success = succes_threshold, return a success

    if #failure = len(children) - succes_threshold, return a failure

    What we can return as value and keep sematic Failure/Success:
     - an array of previous result when success
     - last failure when fail

    Args:
        children (List[CallableFunction]): list of Awaitable
        succes_threshold (int): succes threshold value

    Returns:
        (AsyncInnerFunction): an awaitable function.

    Raises:
        (AssertionError): if succes_threshold is invalid
    """
    _succes_threshold = succes_threshold or len(children)
    if not (0 <= _succes_threshold <= len(children)):
        raise AssertionError('succes_threshold')

    failure_threshold = len(children) - _succes_threshold + 1

    _children = [to_async(child) for child in children]

    @node_metadata(properties=['_succes_threshold'])
    async def _sequence():
        success = 0
        failure = 0
        results = []

        for child in _children:
            last_result = await child()
            results.append(last_result)

            if bool(last_result):
                success += 1
                if success == _succes_threshold:
                    # last evaluation is a success
                    return results
            else:
                failure += 1
                if failure == failure_threshold:
                    # last evaluation is a failure
                    return last_result
        # should be never reached
        return FAILURE

    return _sequence

stringify_analyze(target, indent=0, label=None)

Stringify node representation of specified target.

Parameters:

Name Type Description Default
target CallableFunction

async function to analyze.

 required
indent int

level identation (default to zero).

 0
label Optional[str]

label of current node (default None).

 None

Returns:

Type Description
str

a string node representation.
Source code in async_btree/analyze.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def stringify_analyze(target: Node, indent: int = 0, label: Optional[str] = None) -> str:
    """Stringify node representation of specified target.

    Args:
        target (CallableFunction): async function to analyze.
        indent (int): level identation (default to zero).
        label (Optional[str]): label of current node (default None).

    Returns:
        (str): a string node representation.
    """
    _ident = '    '
    _space = f'{_ident * indent} '
    result: str = ''
    if label:
        result += f'{_space}--({label})--> {target.name}:\n'
        _space += f"{_ident}{' ' * len(label)}"
    else:
        result += f'{_space}--> {target.name}:\n'

    for k, v in target.properties:
        result += f'{_space}    {k}: {v}\n'

    for _label, children in target.edges:
        if children:
            for child in children:
                result += stringify_analyze(target=child, indent=indent + 1, label=_label)
    return result