Skip to content

control

async_btree.control

Control function definition.

Attributes

Functions

condition_guard(condition, child)

Run child only if condition is truthy; return SUCCESS without running child if condition is falsy.

Semantic shorthand for decision(condition, success_tree=child) with no failure tree.

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated as a boolean.

 required
child CallableFunction

callable run only when condition is truthy.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns child result when condition is truthy, or SUCCESS when condition is falsy.
Source code in async_btree/control.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def condition_guard(condition: CallableFunction, child: CallableFunction) -> AsyncInnerFunction:
    """Run child only if condition is truthy; return SUCCESS without running child if condition is falsy.

    Semantic shorthand for `decision(condition, success_tree=child)` with no failure tree.

    Args:
        condition (CallableFunction): sync or async callable evaluated as a boolean.
        child (CallableFunction): callable run only when condition is truthy.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns child result when condition
            is truthy, or `SUCCESS` when condition is falsy.
    """
    return alias_node_metadata(name="condition_guard", target=decision(condition=condition, success_tree=child))

decision(condition, success_tree, failure_tree=None)

Create a decision node.

If condition is truthy, return evaluation of success_tree. Otherwise, return evaluation of failure_tree if set, or SUCCESS.

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated as a boolean.

 required
success_tree CallableFunction

callable evaluated when condition is truthy.

 required
failure_tree CallableFunction | None

callable evaluated when condition is falsy. Defaults to None; when absent, returns SUCCESS.

 None

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
147
148
149
150
151
152
153
154
def decision(
    condition: CallableFunction,
    success_tree: CallableFunction,
    failure_tree: CallableFunction | None = None,
) -> AsyncInnerFunction:
    """Create a decision node.

    If condition is truthy, return evaluation of `success_tree`.
    Otherwise, return evaluation of `failure_tree` if set, or `SUCCESS`.

    Args:
        condition (CallableFunction): sync or async callable evaluated as a boolean.
        success_tree (CallableFunction): callable evaluated when condition is truthy.
        failure_tree (CallableFunction | None): callable evaluated when condition is falsy.
            Defaults to `None`; when absent, returns `SUCCESS`.

    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

do_while(child, condition)

Run child at least once, then repeat while condition is truthy.

Unlike repeat_while, the child always executes at least once before the condition is checked.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable executed each iteration.

 required
condition CallableFunction

sync or async callable checked after each iteration.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the last child result.
Source code in async_btree/control.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def do_while(child: CallableFunction, condition: CallableFunction) -> AsyncInnerFunction:
    """Run child at least once, then repeat while condition is truthy.

    Unlike `repeat_while`, the child always executes at least once before the condition
    is checked.

    Args:
        child (CallableFunction): sync or async callable executed each iteration.
        condition (CallableFunction): sync or async callable checked after each iteration.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the last child result.
    """
    _child = to_async(child)
    _condition = to_async(condition)

    @node_metadata(edges=["_child", "_condition"])
    async def _do_while():
        result: Any = FAILURE
        while True:
            result = await _child()
            if not bool(await _condition()):
                break
        return result

    return _do_while

fallback(children)

Execute children in sequence and succeed as soon as one succeeds; fail if all fail.

Children are evaluated in order from highest to lowest priority. Evaluation stops at the first truthy result. selector is an alias for this function.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the first truthy result, or the last falsy result if all children fail.
Source code in async_btree/control.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def fallback(children: list[CallableFunction]) -> AsyncInnerFunction:
    """Execute children in sequence and succeed as soon as one succeeds; fail if all fail.

    Children are evaluated in order from highest to lowest priority. Evaluation stops
    at the first truthy result. `selector` is an alias for this function.

    Args:
        children (list[CallableFunction]): list of sync or async callables.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the first truthy result,
            or the last falsy result if all children fail.
    """
    return alias_node_metadata(
        name="fallback",
        target=sequence(children, success_threshold=min(1, len(children))),
    )

random_selector(children)

Execute children in a random order, succeeding as soon as one succeeds.

Children are reshuffled on every call, so the evaluation order differs each tick.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns [result] for the first truthy child, or FAILURE if all children fail.
Source code in async_btree/control.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def random_selector(children: list[CallableFunction]) -> AsyncInnerFunction:
    """Execute children in a random order, succeeding as soon as one succeeds.

    Children are reshuffled on every call, so the evaluation order differs each tick.

    Args:
        children (list[CallableFunction]): list of sync or async callables.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns `[result]` for the first
            truthy child, or `FAILURE` if all children fail.
    """
    _children = [to_async(child) for child in children]

    @node_metadata()
    async def _random_selector():
        for child in random.sample(_children, len(_children)):
            result = await child()
            if bool(result):
                return [result]
        return FAILURE

    return _random_selector

repeat_n(child, n)

Run child exactly n times regardless of its result.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to run.

 required
n int

number of times to execute child.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the last child result, or FAILURE if n is 0.
Source code in async_btree/control.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def repeat_n(child: CallableFunction, n: int) -> AsyncInnerFunction:
    """Run child exactly `n` times regardless of its result.

    Args:
        child (CallableFunction): sync or async callable to run.
        n (int): number of times to execute child.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the last child result,
            or `FAILURE` if `n` is 0.
    """
    _child = to_async(child)

    @node_metadata(properties=["n"])
    async def _repeat_n():
        result: Any = FAILURE
        for _ in range(n):
            result = await _child()
        return result

    return _repeat_n

repeat_until(condition, child)

Repeat child until condition becomes truthy, stopping when condition is met.

Returns the last child evaluation, or FAILURE if condition is truthy on the first check (no iteration occurs).

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated before each iteration.

 required
child CallableFunction

sync or async callable executed each iteration.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def repeat_until(condition: CallableFunction, child: CallableFunction) -> AsyncInnerFunction:
    """Repeat child until condition becomes truthy, stopping when condition is met.

    Returns the last child evaluation, or `FAILURE` if condition is truthy on the first check
    (no iteration occurs).

    Args:
        condition (CallableFunction): sync or async callable evaluated before each iteration.
        child (CallableFunction): sync or async callable executed each iteration.

    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 not bool(await _condition()):
            result = await _child()

        return result

    return _repeat_until

repeat_while(condition, child)

Repeat child while condition is truthy, stopping when condition becomes falsy.

Returns the last child evaluation, or FAILURE if condition is falsy on the first check.

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated before each iteration.

 required
child CallableFunction

sync or async callable executed each iteration.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def repeat_while(condition: CallableFunction, child: CallableFunction) -> AsyncInnerFunction:
    """Repeat child while condition is truthy, stopping when condition becomes falsy.

    Returns the last child evaluation, or `FAILURE` if condition is falsy on the first check.

    Args:
        condition (CallableFunction): sync or async callable evaluated before each iteration.
        child (CallableFunction): sync or async callable executed each iteration.

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

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

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

        return result

    return _repeat_while

selector(children)

Alias of fallback. Execute children in sequence and succeed as soon as one succeeds.

Children are evaluated in order from highest to lowest priority. Evaluation stops at the first truthy result.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the first truthy result, or the last falsy result if all children fail.
Source code in async_btree/control.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def selector(children: list[CallableFunction]) -> AsyncInnerFunction:
    """Alias of `fallback`. Execute children in sequence and succeed as soon as one succeeds.

    Children are evaluated in order from highest to lowest priority. Evaluation stops
    at the first truthy result.

    Args:
        children (list[CallableFunction]): list of sync or async callables.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the first truthy result,
            or the last falsy result if all children fail.
    """
    return alias_node_metadata(
        name="selector",
        target=sequence(children, success_threshold=min(1, len(children))),
    )

sequence(children, success_threshold=None)

Return a function which executes children in sequence.

success_threshold generalizes traditional sequence/fallback behaviour and must be in [0, len(children)]. Defaults to None, which means len(children) (all children must succeed).

  • If #success reaches success_threshold, returns the list of results collected so far.
  • If #failure reaches len(children) - success_threshold + 1, returns the last falsy result.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required
success_threshold int | None

minimum number of children that must succeed. Defaults to None (equivalent to len(children)).

 None

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns a list of results on success, or the last falsy result on failure.

Raises:

Type Description
AssertionError

if success_threshold is outside [0, len(children)].
Source code in async_btree/control.py 
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
def sequence(children: list[CallableFunction], success_threshold: int | None = None) -> AsyncInnerFunction:
    """Return a function which executes children in sequence.

    `success_threshold` generalizes traditional sequence/fallback behaviour and
    must be in `[0, len(children)]`. Defaults to `None`, which means `len(children)`
    (all children must succeed).

    - If #success reaches `success_threshold`, returns the list of results collected so far.
    - If #failure reaches `len(children) - success_threshold + 1`, returns the last falsy result.

    Args:
        children (list[CallableFunction]): list of sync or async callables.
        success_threshold (int | None): minimum number of children that must succeed.
            Defaults to `None` (equivalent to `len(children)`).

    Returns:
        (AsyncInnerFunction): an awaitable function that returns a list of results on
            success, or the last falsy result on failure.

    Raises:
        AssertionError: if `success_threshold` is outside `[0, len(children)]`.
    """
    _success_threshold = success_threshold if success_threshold is not None else len(children)
    if not (0 <= _success_threshold <= len(children)):
        raise AssertionError("success_threshold")

    failure_threshold = len(children) - _success_threshold + 1

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

    @node_metadata(properties=["_success_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 == _success_threshold:
                    return results
            else:
                failure += 1
                if failure == failure_threshold:
                    return last_result
        return FAILURE

    return _sequence

switch(condition, cases, default=None)

Route to a child based on the return value of condition.

Evaluates condition to get a key, looks it up in cases, and runs the matching child. Falls back to default if no case matches. Returns FAILURE if no case matches and no default is provided.

The case keys are shown in stringify_analyze output. The children themselves are not traversed by analyze().

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable that returns a hashable key.

 required
cases dict[Any, CallableFunction]

mapping of key → child callable.

 required
default CallableFunction | None

callable run when no case matches. Defaults to None.

 None

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the matched child's result, or FAILURE if no match and no default.
Source code in async_btree/control.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def switch(
    condition: CallableFunction,
    cases: dict[Any, CallableFunction],
    default: CallableFunction | None = None,
) -> AsyncInnerFunction:
    """Route to a child based on the return value of `condition`.

    Evaluates `condition` to get a key, looks it up in `cases`, and runs the
    matching child. Falls back to `default` if no case matches. Returns `FAILURE`
    if no case matches and no default is provided.

    The case keys are shown in `stringify_analyze` output. The children themselves
    are not traversed by `analyze()`.

    Args:
        condition (CallableFunction): sync or async callable that returns a hashable key.
        cases (dict[Any, CallableFunction]): mapping of key → child callable.
        default (CallableFunction | None): callable run when no case matches. Defaults to `None`.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the matched child's result,
            or `FAILURE` if no match and no default.
    """
    _condition = to_async(condition)
    _cases = {k: to_async(v) for k, v in cases.items()}
    _default = to_async(default) if default else None
    _case_keys = list(cases.keys())

    @node_metadata(properties=["_case_keys"], edges=["_default"])
    async def _switch():
        _ = _case_keys  # ensure captured in closure for analyze()
        key = await _condition()
        child = _cases.get(key, _default)
        if child is None:
            return FAILURE
        return await child()

    return _switch