nodes

count_source(limit: int = 10000, start: int = 0) → Generator[Annotated[int, Name(name=index)], None, None][source]
letter_source() → Generator[Annotated[str, Name(name=letter)]][source]
word_source() → Generator[Annotated[str, Name(name=word)]][source]
multi_words_source(n: int) → Generator[Annotated[list[str], Name(name=multi_words)]][source]
sporadic_word(every: int = 3) → Generator[Annotated[str, Name(name=word)] | None, None, None][source]
word_counts() → Generator[tuple[Annotated[str, Name(name=word)], Annotated[list[int], Name(name=counts)]]][source]
multiply(left: int, right: int = 2) → int[source]

Return value purposely unnamed, to be used as {nodename}.value

divide(numerator: int, denominator: int = 5) → Annotated[float, Name(name=ratio)][source]
concat(strings: list[str]) → str[source]
multi_concat(**kwargs: list[str]) → str[source]
exclaim(string: str, hype: int = 1) → str[source]
repeat(string: str, times: int) → str[source]
dictify(key: str, items: list[Any]) → dict[str, Any][source]
error(value: Any) → None[source]
pydantic model CountSource[source]
Config:

  • extra: str = forbid

Fields:
field limit: int = 1000
field start: int = 0
model_post_init(_Node__context: Any) → None

See docstring of process() for description of post init wrapping of generators

process() → Generator[Annotated[int, Name(name=index)], None, None][source]
pydantic model UnannotatedGenerator[source]
Config:

  • extra: str = forbid

Fields:
field limit: int = 1000
field start: int = 0
model_post_init(_Node__context: Any) → None

See docstring of process() for description of post init wrapping of generators

process()[source]
pydantic model Multiply[source]
Config:

  • extra: str = forbid

Fields:

model_post_init(_Node__context: Any) → None

See docstring of process() for description of post init wrapping of generators

process(left: int, right: int = 2) → Annotated[int, Name(name=product)][source]
class VolumeProcess(height: int = 2)[source]
process(width: int, depth: int) → Annotated[int, Name(name=volume)][source]
class Volume(height: int = 2)[source]
volume(width: int, depth: int) → Annotated[int, Name(name=volume)][source]
class Now[source]
print(prefix: str = 'Now: ') → Annotated[str, Name(name=timestamp)][source]
class CountSourceDecor(start: int = 0)[source]
process() → Generator[Annotated[int, Name(name=count)], None, None][source]
input_party(one: int, two: float, three: str, four: bool, five: list, six: dict, seven: set) → Annotated[bool, Name(name=works)][source]
long_add(value: float, sleep_for: float = 0.25) → float[source]
async number_to_letter(number: int, offset: int = 0) → str[source]
class NumberToLetterCls(offset: int = 0)[source]
async process(number: int) → str[source]
async async_error(value: Any) → None[source]

Just raise an error!

class StatefulMultiply(start: int = 0)[source]
process(left: float, right: float = 1) → float[source]
fast_forward(generator: count, n: int = 1) → tuple[Annotated[int, Name(name=next)]][source]
jump(generator: count, n: int = 1) → tuple[Annotated[count, Name(name=skirttt)], Annotated[int, Name(name=next)]][source]
rewind(generator: count, n: int = 1) → Annotated[count, Name(name=skrittt)][source]

Purposely designed to mutate the input and return a new object

zip_iter(*args: Iterator) → tuple[Any, ...][source]
increment(iterator: Iterator[int], increment: int = 1) → tuple[Annotated[Iterator[int], Name(name=iterator)], Annotated[int, Name(name=value)]][source]
passthrough(value: Any, *args: Any, **kwargs: Any) → Any[source]
pydantic model InitCounter[source]

Count how many times we have been initialized and deinitalized

Config:

  • extra: str = forbid

Fields:

deinit() → None[source]
init() → None[source]
model_post_init(_Node__context: Any) → None

See docstring of process() for description of post init wrapping of generators

process() → tuple[Annotated[int, Name(name=inits)], Annotated[int, Name(name=deinits)]][source]
inject_epoch(epoch: Epoch) → Epoch[source]
inject_eventmap(special_value: Any, events: dict[Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)], Event]) → dict[Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)], Event][source]
just_wait(value: Any, wait_for: float = 0.5) → Any[source]
switch() → Generator[tuple[Annotated[str, Name(name=fruits)], Annotated[str, Name(name=vegetables)], Annotated[str, Name(name=minerals)]]][source]

Yield in a cycle from different signals, noeventing in the others

this_or_that(this: Any | None = None, that: Any | None = None, the_other: Any | None = None) → dict[source]

Optional inputs!