22162720669

Committed 19 Feb 2026 12:01AM UTC coverage: 88.173% (+0.8%) from 87.415%

Build # 22162720669

Build Type

Pull #138

github

Committed by

web-flow

Commit Message

Merge 6647f55fd into 338726812

Pull Request Pull Request #138: Implement Map

Coverage Stats

440 of 476 new or added lines in 18 files covered. (92.44%)

5 existing lines in 2 files now uncovered.

3094 of 3509 relevant lines covered (88.17%)

6.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.52

/src/noob/node/base.py

import functools
import inspect
from collections.abc import Callable, Generator, Mapping
from types import GeneratorType, GenericAlias, UnionType
from typing import (  # type: ignore[attr-defined]
    TYPE_CHECKING,
    Annotated,
    Any,
    Self,
    TypeVar,
    Union,
    _UnionGenericAlias,
    cast,
    get_args,
    get_origin,
)

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    ModelWrapValidatorHandler,
    PrivateAttr,
    model_validator,
)

from noob.introspection import is_optional, is_union
from noob.node.spec import NodeSpecification
from noob.types import Epoch, EventMap
from noob.utils import iscoroutinefunction_partial, resolve_python_identifier

if TYPE_CHECKING:
    from noob.input import InputCollection

_INJECTION_MAP = {"epoch": Epoch, "events": EventMap}
"""
Mapping between the keys for things that can be injected in a Process method
to the types that trigger their injection

epoch - the current epoch
events - see EventMap
"""


class Slot(BaseModel):
    name: str
    annotation: Any
    required: bool = True

    @classmethod
    def from_callable(cls, func: Callable) -> dict[str, "Slot"]:
        slots = {}
        sig = inspect.signature(func)

        for name, param in sig.parameters.items():
            slots[name] = Slot(
                name=name,
                annotation=param.annotation,
                required=not (is_optional(param.annotation) and param.default is None),
            )
        return slots


class Signal(BaseModel):
    name: str
    type_: type | None | UnionType | GenericAlias | _UnionGenericAlias

    # Unable to generate pydantic-core schema for <class 'types.UnionType'>
    model_config = ConfigDict(arbitrary_types_allowed=True)

    @classmethod
    def from_callable(cls, func: Callable) -> list["Signal"]:
        signals = []
        return_annotation = inspect.signature(func).return_annotation
        for name, type_ in cls._collect_signal_names(return_annotation):
            signals.append(Signal(name=name, type_=type_))
        return signals

    @classmethod
    def _collect_signal_names(
        cls, return_annotation: Annotated[Any, Any]
    ) -> list[tuple[str, type]]:
        """
        Recursive kernel for extracting name attribute of Name metadata from a type annotation.
        If the outermost origin is `typing.Annotated`, it extracts all Name objects from its
        arguments and append.
        If the outermost origin is `tuple` or `Generator`, grab the argument and recurses into
        this function.
        If the outermost is `Generator`, or `tuple` but the inner layer isn't one of `Generator`,
        `tuple`, or `Annotated`, assume it's an unnamed signal (no Name inside).
        If the outermost is none of `Generator`, `tuple`, or `Annotated`, assume it's an unnamed
        signal.

        Returns a list of names.
        """
        from noob import Name

        default_name = "value"
        names = []

        # --- get yield type of generators before handling other types
        if get_origin(return_annotation) is Generator:
            return_annotation = get_args(return_annotation)[0]
        # ---

        # def func() -> Annotated[...]
        if get_origin(return_annotation) is Annotated:
            for argument in get_args(return_annotation):
                if isinstance(argument, Name):
                    names.append((argument.name, get_args(return_annotation)[0]))

        # def func() -> tuple[...]: OR def func() -> A | B:
        elif get_origin(return_annotation) is tuple or is_union(return_annotation):
            for argument in get_args(return_annotation):
                if get_origin(argument) in [Annotated, tuple]:
                    names += Signal._collect_signal_names(argument)
                elif argument not in [type(None), None]:
                    names = [(default_name, return_annotation)]

        # def func() -> type:
        elif return_annotation and (return_annotation is not inspect.Signature.empty):
            names = [(default_name, return_annotation)]

        return names


class Edge(BaseModel):
    """
    Directed connection between an output slot a node and an input slot in another node
    """

    source_node: str
    source_signal: str
    target_node: str
    target_slot: str | int | None = None
    """
    - For kwargs, target_slot is the name of the kwarg that the value is passed to.
    - For positional arguments, target_slot is an integer that indicates the index of the arg
    - For scalar arguments, target slot is None 
    """
    required: bool = True


_PROCESS_METHOD_SENTINEL = "__is_process_method__"
_GENERATOR_METHOD_SENTINEL = "__is_generator_method__"

_TProcess = TypeVar("_TProcess", bound=Callable | Generator)


def process_method(func: _TProcess) -> _TProcess:
    """
    Decorator to mark a method as the designated 'process' method for a class.
    """
    setattr(func, _PROCESS_METHOD_SENTINEL, True)
    return func


class Node(BaseModel):
    """A node within a processing tube"""

    id: str
    """Unique identifier of the node"""
    spec: NodeSpecification | None = None
    enabled: bool = True
    """Starting state for a node being enabled. When a node is disabled, 
    it will be deinitialized and removed from the processing graph, 
    but the node object will still be kept by the Tube. Nodes can be disabled 
    and enabled during a tube's operation without recreating the tube. 
    When a node is disabled, other nodes that depend on it will not be disabled, 
    but they may never be called since their dependencies will never be satisfied."""
    stateful: bool = True
    """
    Whether this node is stateful (True), or stateless (False).
    Stateful nodes are assumed to care about the order in which they receive events - 
    i.e. for a given set of inputs, the values returned by ``process`` are different
    when called in a different order.
    
    This attribute has no effect in synchronous runners,
    but in concurrent runners where multiple epochs of events can be processed simultaneously,
    setting a node as stateless can improve performance
    as the node processes events as soon as it receives them rather than waiting
    for the next epoch in the sequence to arrive.
    
    Defined as an instance, 
    rather than a class attribute to allow it being overridden by a node specification.
    Subclasses should override the default value to be considered stateless by default.
    
    By default, unless specified otherwise:
    
    * Class nodes are considered stateful
    * Generator nodes are considered stateful
    * Function nodes are considered stateless
    """

    _signals: list[Signal] | None = None
    _slots: dict[str, Slot] | None = None
    _gen: Generator | None = None
    _edges: list[Edge] | None = None
    _injections: dict[str, str] | None = None

    model_config = ConfigDict(extra="forbid")

    def model_post_init(self, __context: Any) -> None:
        """See docstring of :meth:`.process` for description of post init wrapping of generators"""
        if inspect.isgeneratorfunction(self.process):
            self._wrap_generator(self.process)

    # TODO: Support dependency injection in mypy plugin
    def init(self) -> None:
        """
        Start producing, processing, or receiving data.

        Default is a no-op.
        Subclasses do not need to override if they have no initialization logic.

        Subclasses MAY add a `context: RunnerContext` param to request information
        about the enclosing runner while initializing
        """
        pass

    def deinit(self) -> None:
        """
        Stop producing, processing, or receiving data

        Default is a no-op.
        Subclasses do not need to override if they have no deinit logic.
        """
        pass

    def process(self, *args: Any, **kwargs: Any) -> Any | None:
        """
        Process some input, emitting it. See subclasses for details.

        If the process method is a generator, when the Node class is instantiated,
        this method is replaced by one that wraps creating and calling the generator.

        something like this:

        ```python
        gen = self.process()
        self.process = lambda: next(gen)
        ```

        Note that `send` handling is not implemented for generators,
        so `process` methods that are generators cannot depend on events from any other nodes
        (i.e. behave like source nodes).
        """
        raise NotImplementedError()

    @classmethod
    def from_specification(
        cls, spec: "NodeSpecification", input_collection: Union["InputCollection", None] = None
    ) -> "Node":
        """
        Create a node from its spec

        - resolve the node type
        - if a function, wrap it in a node class
        - if a class, just instantiate it
        """
        obj = resolve_python_identifier(spec.type_)

        params = spec.params if spec.params is not None else {}
        if input_collection:
            params = input_collection.get_node_params(params)
        elif params:
            from noob.input import InputCollection

            if any(
                InputCollection.INPUT_PATTERN.fullmatch(v)
                for v in params.values()
                if isinstance(v, str)
            ):
                raise ValueError("No input collection supplied, but inputs specified in params")

        # additional kwargs that can be present or absent without default
        kwargs = {}
        if spec.stateful is not None:
            kwargs["stateful"] = spec.stateful

        # check if function by checking if callable -
        # Node classes do not have __call__ defined and thus should not be callable
        if inspect.isclass(obj):
            if issubclass(obj, Node):
                return obj(id=spec.id, spec=spec, enabled=spec.enabled, **params, **kwargs)
            else:
                return WrapClassNode(
                    id=spec.id, cls=obj, spec=spec, params=params, enabled=spec.enabled, **kwargs
                )
        else:
            return WrapFuncNode(
                id=spec.id, fn=obj, spec=spec, params=params, enabled=spec.enabled, **kwargs
            )

    @property
    def signals(self) -> list[Signal]:
        if self._signals is None:
            self._signals = self._collect_signals()
        if not self._signals:
            self._signals = [Signal(name="value", type_=Any)]
        return self._signals

    def _collect_signals(self) -> list[Signal]:
        return Signal.from_callable(self.process)

    @property
    def slots(self) -> dict[str, Slot]:
        if self._slots is None:
            self._slots = self._collect_slots()
        return self._slots

    @property
    def edges(self) -> list[Edge]:
        """
        The dependencies this node has declared, express as edges between
        another node's signals and our slots
        """
        if not self.spec:
            raise ValueError("Node has no dependency specification, edges are undefined")
        if not self.spec.depends:
            return []

        if self._edges is not None:
            return self._edges

        edges = []
        if isinstance(self.spec.depends, str):
            # handle scalar dependency like
            # depends: node.slot
            source_node, source_signal = self.spec.depends.split(".")
            edges.append(
                Edge(
                    source_node=source_node,
                    source_signal=source_signal,
                    target_node=self.id,
                    target_slot=None,
                )
            )
        else:
            # handle arrays of dependencies, positional and kwargs
            target_slot: int | str
            position_index = 0
            for arrow in self.spec.depends:
                required = True
                if isinstance(arrow, Mapping):  # keyword argument
                    target_slot, source_signal = next(iter(arrow.items()))
                    if target_slot in self.slots:
                        required = self.slots[target_slot].required

                elif isinstance(arrow, str):  # positional argument
                    target_slot = position_index
                    source_signal = arrow
                    position_index += 1

                else:
                    raise NotImplementedError(
                        "Only supporting signal-slot mapping or node pointer."
                    )

                source_node, source_signal = source_signal.split(".")

                edges.append(
                    Edge(
                        source_node=source_node,
                        source_signal=source_signal,
                        target_node=self.id,
                        target_slot=target_slot,
                        required=required,
                    )
                )

        self._edges = edges
        return self._edges

    @functools.cached_property
    def injections(self) -> dict[str, str]:
        """
        If a node's process method requests a dependency to be injected,
        returns a map from the type of inejction to the kwargs to pass them as.
        """
        sig = inspect.signature(self.process)
        injections = {}
        for injection_key, inj_type in _INJECTION_MAP.items():
            for param, param_info in sig.parameters.items():
                if not param_info.annotation:
                    continue
                if param_info.annotation is inj_type:
                    injections[injection_key] = param

        return injections

    @functools.cached_property
    def is_coroutine(self) -> bool:
        """
        is the process method a coroutine?

        (checking this on every call proves to be surprisingly expensive)
        """
        return iscoroutinefunction_partial(self.process)

    def _collect_slots(self) -> dict[str, Slot]:
        return Slot.from_callable(self.process)

    def _wrap_generator(self, proc: Callable[[], GeneratorType]) -> None:
        """
        Wrap a `process` method when it is a generator,
        invoked in `model_post_init`
        """
        self._gen = proc()

        def _process():  # noqa: ANN202
            self._gen = cast(Generator, self._gen)
            return next(self._gen)

        signature = inspect.signature(self.process)

        args = get_args(signature.return_annotation)
        if args:
            _process.__annotations__ = {"return": args[0]}

        self.__dict__["process"] = _process


class WrapClassNode(Node):
    """
    Wrap a non-Node class that has annotated one of its methods as being the "process" method
    using the :func:`process_method` decorator.

    Wrapping allows us to use arbitrary classes as Nodes within noob,
    which expects a `process` method,
    but avoids the problem of potentially breaking the class if it has its own attribute or method
    named `process`.

    After instantiating the outer wrapping class,
    instantiate the inner wrapped class using the `params` given to the outer wrapping class
    during :meth:`.model_post_init` .
    Then dynamically assign the discovered process method on the inner class to the
    outer class as `process`.

    Dynamic discovery at instantiation time, rather than statically defining an outer
    `process` method that then calls the inner method annotated with `process_method`
    does two things:

    - Allows us to statically infer whether the method is a regular function that `return`s or
      a generator using :func:`inspect.isgeneratorfunction` , which relies on a flag
      set on a method at the time it is defined: e.g. a method that internally switches between
      `return self._wrapped()` and `yield from self._wrapped()` would not be correctly detected.
    - Avoids modifying the signature of the wrapped process method with generic args and kwargs
    """

    cls: type
    params: dict[str, Any] = Field(default_factory=dict)

    instance: type | None = None
    _gen: Generator | None = PrivateAttr(default=None)

    def model_post_init(self, context: Any, /) -> None:
        """
        Get the method decorated with :func:`.process_method`,
        assign it to `process`, see class docstring.
        """
        self.instance = self.cls(**self.params)
        fn_name = self._get_process_method(self.cls)
        fn = getattr(self.instance, fn_name)
        self.__dict__["process"] = fn
        super().model_post_init(context)

    def deinit(self) -> None:
        self.instance = None

    def _collect_signals(self) -> list[Signal]:
        return Signal.from_callable(self.process)

    def _collect_slots(self) -> dict[str, Slot]:
        return Slot.from_callable(self.process)

    def _get_process_method(self, cls: type) -> str:
        process_func = None
        for name, member in inspect.getmembers(cls, predicate=inspect.isfunction):
            if hasattr(member, _PROCESS_METHOD_SENTINEL):
                if process_func:
                    raise TypeError(
                        f"Class {cls.__name__} has multiple 'process' methods. Only one is allowed."
                    )
                process_func = name
        if process_func is None:
            if hasattr(cls, "process") and inspect.isfunction(cls.process):
                process_func = "process"
            else:
                raise TypeError(
                    "Class must have 'process' method or decorate a method with @process_method."
                )

        return process_func


class WrapFuncNode(Node):
    fn: Callable
    params: dict = Field(default_factory=dict)
    stateful: bool = False
    """
    Function nodes are considered stateless by default,
    except if they are generators, which are typically stateful. 
    """

    def model_post_init(self, __context: Any) -> None:
        """
        Complete wrapping `fn` without calling `super()`
        because we need to pass `params` to the function if it is a generator,
        and create a :func:`functools.partial` of it if it is not.
        """
        if inspect.isgeneratorfunction(self.fn):
            self._gen = self.fn(**self.params)
            self.__dict__["process"] = lambda: next(self._gen)
        elif inspect.isasyncgenfunction(self.fn):
            raise NotImplementedError("async generators not supported")
        else:
            self.__dict__["process"] = functools.partial(self.fn, **self.params)

    @model_validator(mode="wrap")
    @classmethod
    def set_default_statefulness(cls, data: Any, handler: ModelWrapValidatorHandler[Self]) -> Self:
        """
        If no `stateful` argument is provided explicitly,
        set stateful default False for functions and True for generators
        """
        statefulness_set = isinstance(data, dict) and data.get("stateful", None) is not None
        value = handler(data)
        if statefulness_set:
            return value

        value.stateful = bool(inspect.isgeneratorfunction(value.fn))
        return value

    def _collect_signals(self) -> list[Signal]:
        return Signal.from_callable(self.fn)

    def _collect_slots(self) -> dict[str, Slot]:
        return Slot.from_callable(self.fn)

1	import functools	8✔
2	import inspect	8✔
3	from collections.abc import Callable, Generator, Mapping	8✔
4	from types import GeneratorType, GenericAlias, UnionType	8✔
5	from typing import ( # type: ignore[attr-defined]	8✔
6	TYPE_CHECKING,
7	Annotated,
8	Any,
9	Self,
10	TypeVar,
11	Union,
12	_UnionGenericAlias,
13	cast,
14	get_args,
15	get_origin,
16	)
17
18	from pydantic import (	8✔
19	BaseModel,
20	ConfigDict,
21	Field,
22	ModelWrapValidatorHandler,
23	PrivateAttr,
24	model_validator,
25	)
26
27	from noob.introspection import is_optional, is_union	8✔
28	from noob.node.spec import NodeSpecification	8✔
29	from noob.types import Epoch, EventMap	8✔
30	from noob.utils import iscoroutinefunction_partial, resolve_python_identifier	8✔
31
32	if TYPE_CHECKING:
33	from noob.input import InputCollection
34
35	_INJECTION_MAP = {"epoch": Epoch, "events": EventMap}	8✔
36	"""	8✔
37	Mapping between the keys for things that can be injected in a Process method
38	to the types that trigger their injection
39
40	epoch - the current epoch
41	events - see EventMap
42	"""
43
44
45	class Slot(BaseModel):	8✔
46	name: str	8✔
47	annotation: Any	8✔
48	required: bool = True	8✔
49
50	@classmethod	8✔
51	def from_callable(cls, func: Callable) -> dict[str, "Slot"]:	8✔
52	slots = {}	8✔
53	sig = inspect.signature(func)	8✔
54
55	for name, param in sig.parameters.items():	8✔
56	slots[name] = Slot(	8✔
57	name=name,
58	annotation=param.annotation,
59	required=not (is_optional(param.annotation) and param.default is None),
60	)
61	return slots	8✔
62
63
64	class Signal(BaseModel):	8✔
65	name: str	8✔
66	type_: type \| None \| UnionType \| GenericAlias \| _UnionGenericAlias	8✔
67
68	# Unable to generate pydantic-core schema for <class 'types.UnionType'>
69	model_config = ConfigDict(arbitrary_types_allowed=True)	8✔
70
71	@classmethod	8✔
72	def from_callable(cls, func: Callable) -> list["Signal"]:	8✔
73	signals = []	8✔
74	return_annotation = inspect.signature(func).return_annotation	8✔
75	for name, type_ in cls._collect_signal_names(return_annotation):	8✔
76	signals.append(Signal(name=name, type_=type_))	8✔
77	return signals	8✔
78
79	@classmethod	8✔
80	def _collect_signal_names(	8✔
81	cls, return_annotation: Annotated[Any, Any]
82	) -> list[tuple[str, type]]:
83	"""
84	Recursive kernel for extracting name attribute of Name metadata from a type annotation.
85	If the outermost origin is `typing.Annotated`, it extracts all Name objects from its
86	arguments and append.
87	If the outermost origin is `tuple` or `Generator`, grab the argument and recurses into
88	this function.
89	If the outermost is `Generator`, or `tuple` but the inner layer isn't one of `Generator`,
90	`tuple`, or `Annotated`, assume it's an unnamed signal (no Name inside).
91	If the outermost is none of `Generator`, `tuple`, or `Annotated`, assume it's an unnamed
92	signal.
93
94	Returns a list of names.
95	"""
96	from noob import Name	8✔
97
98	default_name = "value"	8✔
99	names = []	8✔
100
101	# --- get yield type of generators before handling other types
102	if get_origin(return_annotation) is Generator:	8✔
103	return_annotation = get_args(return_annotation)[0]	8✔
104	# ---
105
106	# def func() -> Annotated[...]
107	if get_origin(return_annotation) is Annotated:	8✔
108	for argument in get_args(return_annotation):	8✔
109	if isinstance(argument, Name):	8✔
110	names.append((argument.name, get_args(return_annotation)[0]))	8✔
111
112	# def func() -> tuple[...]: OR def func() -> A \| B:
113	elif get_origin(return_annotation) is tuple or is_union(return_annotation):	8✔
114	for argument in get_args(return_annotation):	8✔
115	if get_origin(argument) in [Annotated, tuple]:	8✔
116	names += Signal._collect_signal_names(argument)	8✔
117	elif argument not in [type(None), None]:	8✔
118	names = [(default_name, return_annotation)]	8✔
119
120	# def func() -> type:
121	elif return_annotation and (return_annotation is not inspect.Signature.empty):	8✔
122	names = [(default_name, return_annotation)]	8✔
123
124	return names	8✔
125
126
127	class Edge(BaseModel):	8✔
128	"""
129	Directed connection between an output slot a node and an input slot in another node
130	"""
131
132	source_node: str	8✔
133	source_signal: str	8✔
134	target_node: str	8✔
135	target_slot: str \| int \| None = None	8✔
136	"""	8✔
137	- For kwargs, target_slot is the name of the kwarg that the value is passed to.
138	- For positional arguments, target_slot is an integer that indicates the index of the arg
139	- For scalar arguments, target slot is None
140	"""
141	required: bool = True	8✔
142
143
144	_PROCESS_METHOD_SENTINEL = "__is_process_method__"	8✔
145	_GENERATOR_METHOD_SENTINEL = "__is_generator_method__"	8✔
146
147	_TProcess = TypeVar("_TProcess", bound=Callable \| Generator)	8✔
148
149
150	def process_method(func: _TProcess) -> _TProcess:	8✔
151	"""
152	Decorator to mark a method as the designated 'process' method for a class.
153	"""
154	setattr(func, _PROCESS_METHOD_SENTINEL, True)	8✔
155	return func	8✔
156
157
158	class Node(BaseModel):	8✔
159	"""A node within a processing tube"""
160
161	id: str	8✔
162	"""Unique identifier of the node"""	8✔
163	spec: NodeSpecification \| None = None	8✔
164	enabled: bool = True	8✔
165	"""Starting state for a node being enabled. When a node is disabled,	8✔
166	it will be deinitialized and removed from the processing graph,
167	but the node object will still be kept by the Tube. Nodes can be disabled
168	and enabled during a tube's operation without recreating the tube.
169	When a node is disabled, other nodes that depend on it will not be disabled,
170	but they may never be called since their dependencies will never be satisfied."""
171	stateful: bool = True	8✔
172	"""	8✔
173	Whether this node is stateful (True), or stateless (False).
174	Stateful nodes are assumed to care about the order in which they receive events -
175	i.e. for a given set of inputs, the values returned by ``process`` are different
176	when called in a different order.
177
178	This attribute has no effect in synchronous runners,
179	but in concurrent runners where multiple epochs of events can be processed simultaneously,
180	setting a node as stateless can improve performance
181	as the node processes events as soon as it receives them rather than waiting
182	for the next epoch in the sequence to arrive.
183
184	Defined as an instance,
185	rather than a class attribute to allow it being overridden by a node specification.
186	Subclasses should override the default value to be considered stateless by default.
187
188	By default, unless specified otherwise:
189
190	* Class nodes are considered stateful
191	* Generator nodes are considered stateful
192	* Function nodes are considered stateless
193	"""
194
195	_signals: list[Signal] \| None = None	8✔
196	_slots: dict[str, Slot] \| None = None	8✔
197	_gen: Generator \| None = None	8✔
198	_edges: list[Edge] \| None = None	8✔
199	_injections: dict[str, str] \| None = None	8✔
200
201	model_config = ConfigDict(extra="forbid")	8✔
202
203	def model_post_init(self, __context: Any) -> None:	8✔
204	"""See docstring of :meth:`.process` for description of post init wrapping of generators"""
205	if inspect.isgeneratorfunction(self.process):	8✔
206	self._wrap_generator(self.process)	8✔
207
208	# TODO: Support dependency injection in mypy plugin
209	def init(self) -> None:	8✔
210	"""
211	Start producing, processing, or receiving data.
212
213	Default is a no-op.
214	Subclasses do not need to override if they have no initialization logic.
215
216	Subclasses MAY add a `context: RunnerContext` param to request information
217	about the enclosing runner while initializing
218	"""
219	pass	8✔
220
221	def deinit(self) -> None:	8✔
222	"""
223	Stop producing, processing, or receiving data
224
225	Default is a no-op.
226	Subclasses do not need to override if they have no deinit logic.
227	"""
228	pass	8✔
229
230	def process(self, args: Any, *kwargs: Any) -> Any \| None:	8✔
231	"""
232	Process some input, emitting it. See subclasses for details.
233
234	If the process method is a generator, when the Node class is instantiated,
235	this method is replaced by one that wraps creating and calling the generator.
236
237	something like this:
238
239	```python
240	gen = self.process()
241	self.process = lambda: next(gen)
242	```
243
244	Note that `send` handling is not implemented for generators,
245	so `process` methods that are generators cannot depend on events from any other nodes
246	(i.e. behave like source nodes).
247	"""
248	raise NotImplementedError()	×
249
250	@classmethod	8✔
251	def from_specification(	8✔
252	cls, spec: "NodeSpecification", input_collection: Union["InputCollection", None] = None
253	) -> "Node":
254	"""
255	Create a node from its spec
256
257	- resolve the node type
258	- if a function, wrap it in a node class
259	- if a class, just instantiate it
260	"""
261	obj = resolve_python_identifier(spec.type_)	8✔
262
263	params = spec.params if spec.params is not None else {}	8✔
264	if input_collection:	8✔
265	params = input_collection.get_node_params(params)	8✔
266	elif params:	8✔
267	from noob.input import InputCollection	8✔
268
269	if any(	8✔
270	InputCollection.INPUT_PATTERN.fullmatch(v)
271	for v in params.values()
272	if isinstance(v, str)
273	):
274	raise ValueError("No input collection supplied, but inputs specified in params")	×
275
276	# additional kwargs that can be present or absent without default
277	kwargs = {}	8✔
278	if spec.stateful is not None:	8✔
279	kwargs["stateful"] = spec.stateful	5✔
280
281	# check if function by checking if callable -
282	# Node classes do not have __call__ defined and thus should not be callable
283	if inspect.isclass(obj):	8✔
284	if issubclass(obj, Node):	8✔
285	return obj(id=spec.id, spec=spec, enabled=spec.enabled, params, kwargs)	8✔
286	else:
287	return WrapClassNode(	8✔
288	id=spec.id, cls=obj, spec=spec, params=params, enabled=spec.enabled, **kwargs
289	)
290	else:
291	return WrapFuncNode(	8✔
292	id=spec.id, fn=obj, spec=spec, params=params, enabled=spec.enabled, **kwargs
293	)
294
295	@property	8✔
296	def signals(self) -> list[Signal]:	8✔
297	if self._signals is None:	8✔
298	self._signals = self._collect_signals()	8✔
299	if not self._signals:	8✔
300	self._signals = [Signal(name="value", type_=Any)]	8✔
301	return self._signals	8✔
302
303	def _collect_signals(self) -> list[Signal]:	8✔
304	return Signal.from_callable(self.process)	8✔
305
306	@property	8✔
307	def slots(self) -> dict[str, Slot]:	8✔
308	if self._slots is None:	8✔
309	self._slots = self._collect_slots()	8✔
310	return self._slots	8✔
311
312	@property	8✔
313	def edges(self) -> list[Edge]:	8✔
314	"""
315	The dependencies this node has declared, express as edges between
316	another node's signals and our slots
317	"""
318	if not self.spec:	8✔
319	raise ValueError("Node has no dependency specification, edges are undefined")	×
320	if not self.spec.depends:	8✔
321	return []	8✔
322
323	if self._edges is not None:	8✔
324	return self._edges	8✔
325
326	edges = []	8✔
327	if isinstance(self.spec.depends, str):	8✔
328	# handle scalar dependency like
329	# depends: node.slot
330	source_node, source_signal = self.spec.depends.split(".")	8✔
331	edges.append(	8✔
332	Edge(
333	source_node=source_node,
334	source_signal=source_signal,
335	target_node=self.id,
336	target_slot=None,
337	)
338	)
339	else:
340	# handle arrays of dependencies, positional and kwargs
341	target_slot: int \| str
342	position_index = 0	8✔
343	for arrow in self.spec.depends:	8✔
344	required = True	8✔
345	if isinstance(arrow, Mapping): # keyword argument	8✔
346	target_slot, source_signal = next(iter(arrow.items()))	8✔
347	if target_slot in self.slots:	8✔
348	required = self.slots[target_slot].required	8✔
349
350	elif isinstance(arrow, str): # positional argument	8✔
351	target_slot = position_index	8✔
352	source_signal = arrow	8✔
353	position_index += 1	8✔
354
355	else:
356	raise NotImplementedError(	×
357	"Only supporting signal-slot mapping or node pointer."
358	)
359
360	source_node, source_signal = source_signal.split(".")	8✔
361
362	edges.append(	8✔
363	Edge(
364	source_node=source_node,
365	source_signal=source_signal,
366	target_node=self.id,
367	target_slot=target_slot,
368	required=required,
369	)
370	)
371
372	self._edges = edges	8✔
373	return self._edges	8✔
374
375	@functools.cached_property	8✔
376	def injections(self) -> dict[str, str]:	8✔
377	"""
378	If a node's process method requests a dependency to be injected,
379	returns a map from the type of inejction to the kwargs to pass them as.
380	"""
381	sig = inspect.signature(self.process)	8✔
382	injections = {}	8✔
383	for injection_key, inj_type in _INJECTION_MAP.items():	8✔
384	for param, param_info in sig.parameters.items():	8✔
385	if not param_info.annotation:	8✔
NEW 386	continue	×
387	if param_info.annotation is inj_type:	8✔
388	injections[injection_key] = param	8✔
389
390	return injections	8✔
391
392	@functools.cached_property	8✔
393	def is_coroutine(self) -> bool:	8✔
394	"""
395	is the process method a coroutine?
396
397	(checking this on every call proves to be surprisingly expensive)
398	"""
399	return iscoroutinefunction_partial(self.process)	8✔
400
401	def _collect_slots(self) -> dict[str, Slot]:	8✔
402	return Slot.from_callable(self.process)	8✔
403
404	def _wrap_generator(self, proc: Callable[[], GeneratorType]) -> None:	8✔
405	"""
406	Wrap a `process` method when it is a generator,
407	invoked in `model_post_init`
408	"""
409	self._gen = proc()	8✔
410
411	def _process(): # noqa: ANN202	8✔
412	self._gen = cast(Generator, self._gen)	8✔
413	return next(self._gen)	8✔
414
415	signature = inspect.signature(self.process)	8✔
416
417	args = get_args(signature.return_annotation)	8✔
418	if args:	8✔
419	_process.__annotations__ = {"return": args[0]}	8✔
420
421	self.__dict__["process"] = _process	8✔
422
423
424	class WrapClassNode(Node):	8✔
425	"""
426	Wrap a non-Node class that has annotated one of its methods as being the "process" method
427	using the :func:`process_method` decorator.
428
429	Wrapping allows us to use arbitrary classes as Nodes within noob,
430	which expects a `process` method,
431	but avoids the problem of potentially breaking the class if it has its own attribute or method
432	named `process`.
433
434	After instantiating the outer wrapping class,
435	instantiate the inner wrapped class using the `params` given to the outer wrapping class
436	during :meth:`.model_post_init` .
437	Then dynamically assign the discovered process method on the inner class to the
438	outer class as `process`.
439
440	Dynamic discovery at instantiation time, rather than statically defining an outer
441	`process` method that then calls the inner method annotated with `process_method`
442	does two things:
443
444	- Allows us to statically infer whether the method is a regular function that `return`s or
445	a generator using :func:`inspect.isgeneratorfunction` , which relies on a flag
446	set on a method at the time it is defined: e.g. a method that internally switches between
447	`return self._wrapped()` and `yield from self._wrapped()` would not be correctly detected.
448	- Avoids modifying the signature of the wrapped process method with generic args and kwargs
449	"""
450
451	cls: type	8✔
452	params: dict[str, Any] = Field(default_factory=dict)	8✔
453
454	instance: type \| None = None	8✔
455	_gen: Generator \| None = PrivateAttr(default=None)	8✔
456
457	def model_post_init(self, context: Any, /) -> None:	8✔
458	"""
459	Get the method decorated with :func:`.process_method`,
460	assign it to `process`, see class docstring.
461	"""
462	self.instance = self.cls(**self.params)	8✔
463	fn_name = self._get_process_method(self.cls)	8✔
464	fn = getattr(self.instance, fn_name)	8✔
465	self.__dict__["process"] = fn	8✔
466	super().model_post_init(context)	8✔
467
468	def deinit(self) -> None:	8✔
469	self.instance = None	8✔
470
471	def _collect_signals(self) -> list[Signal]:	8✔
472	return Signal.from_callable(self.process)	8✔
473
474	def _collect_slots(self) -> dict[str, Slot]:	8✔
475	return Slot.from_callable(self.process)	8✔
476
477	def _get_process_method(self, cls: type) -> str:	8✔
478	process_func = None	8✔
479	for name, member in inspect.getmembers(cls, predicate=inspect.isfunction):	8✔
480	if hasattr(member, _PROCESS_METHOD_SENTINEL):	8✔
481	if process_func:	8✔
482	raise TypeError(	×
483	f"Class {cls.__name__} has multiple 'process' methods. Only one is allowed."
484	)
485	process_func = name	8✔
486	if process_func is None:	8✔
487	if hasattr(cls, "process") and inspect.isfunction(cls.process):	8✔
488	process_func = "process"	8✔
489	else:
490	raise TypeError(	×
491	"Class must have 'process' method or decorate a method with @process_method."
492	)
493
494	return process_func	8✔
495
496
497	class WrapFuncNode(Node):	8✔
498	fn: Callable	8✔
499	params: dict = Field(default_factory=dict)	8✔
500	stateful: bool = False	8✔
501	"""	8✔
502	Function nodes are considered stateless by default,
503	except if they are generators, which are typically stateful.
504	"""
505
506	def model_post_init(self, __context: Any) -> None:	8✔
507	"""
508	Complete wrapping `fn` without calling `super()`
509	because we need to pass `params` to the function if it is a generator,
510	and create a :func:`functools.partial` of it if it is not.
511	"""
512	if inspect.isgeneratorfunction(self.fn):	8✔
513	self._gen = self.fn(**self.params)	8✔
514	self.__dict__["process"] = lambda: next(self._gen)	8✔
515	elif inspect.isasyncgenfunction(self.fn):	8✔
516	raise NotImplementedError("async generators not supported")	×
517	else:
518	self.__dict__["process"] = functools.partial(self.fn, **self.params)	8✔
519
520	@model_validator(mode="wrap")	8✔
521	@classmethod	8✔
522	def set_default_statefulness(cls, data: Any, handler: ModelWrapValidatorHandler[Self]) -> Self:	8✔
523	"""
524	If no `stateful` argument is provided explicitly,
525	set stateful default False for functions and True for generators
526	"""
527	statefulness_set = isinstance(data, dict) and data.get("stateful", None) is not None	8✔
528	value = handler(data)	8✔
529	if statefulness_set:	8✔
530	return value	8✔
531
532	value.stateful = bool(inspect.isgeneratorfunction(value.fn))	8✔
533	return value	8✔
534
535	def _collect_signals(self) -> list[Signal]:	8✔
536	return Signal.from_callable(self.fn)	8✔
537
538	def _collect_slots(self) -> dict[str, Slot]:	8✔
539	return Slot.from_callable(self.fn)	8✔

miniscope / noob / 22162720669

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous