### Define Node I/O Metadata with Annotated Types (Python)

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

This Python code snippet illustrates how to define input and output metadata for a funcnodes-core node using `typing.Annotated`. It specifies names, descriptions, default values, and trigger behavior for inputs, and defines the output metadata. The example also demonstrates how to instantiate the node, access its metadata, update input values, and trigger execution.

```python
from typing import Annotated
import funcnodes_core as fn

@fn.NodeDecorator(node_id="annotated.example")
def process_data(
    input_value: Annotated[
        int,
        fn.InputMeta(
            name="Value",
            description="The input value to process",
            default=10,
            does_trigger=True,
            hidden=False,
            value_options={"min": 0, "max": 100}
        )
    ],
    scale: Annotated[
        float,
        fn.InputMeta(
            name="Scale Factor",
            description="Multiplier for the value",
            default=1.0,
            does_trigger=False  # Changing scale doesn't auto-trigger
        )
    ]
) -> Annotated[
    float,
    fn.OutputMeta(
        name="Result",
        description="The processed output value"
    )
]:
    return input_value * scale

node = process_data()
print(node.inputs["input_value"].name)        # "Value"
print(node.inputs["input_value"].does_trigger) # True
print(node.inputs["scale"].does_trigger)       # False

node["input_value"] = 50
node["scale"] = 2.5
# Assuming an async context for await
# await node 
# print(node.outputs["Result"].value)  # 125.0 (if executed)
```

--------------------------------

### Library and Shelf for Node Organization in Python

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Illustrates how to organize nodes into hierarchical structures using Library and Shelf. This includes defining nodes via decorators, creating nested shelves, and retrieving nodes by ID.

```python
from funcnodes_core import Library, Shelf, Node, NodeInput, NodeOutput
import funcnodes_core as fn

# Define node classes
@fn.NodeDecorator(node_id="math.add")
def add(a: float, b: float) -> float:
    return a + b

@fn.NodeDecorator(node_id="math.subtract")
def subtract(a: float, b: float) -> float:
    return a - b

@fn.NodeDecorator(node_id="string.concat")
def concat(a: str, b: str) -> str:
    return a + b

# Create hierarchical shelf structure
math_shelf = Shelf(
    name="Math",
    description="Mathematical operations",
    nodes=[add, subtract],
    subshelves=[
        Shelf(name="Advanced", nodes=[], subshelves=[])
    ]
)

string_shelf = Shelf(
    name="String",
    description="String operations",
    nodes=[concat]
)

# Create library and add shelves
lib = Library()
lib.add_shelf(math_shelf)
lib.add_shelf(string_shelf)

# Find nodes in library
paths = lib.find_nodeid("math.add")  # Returns [["Math"]]
print(lib.has_node_id("math.add"))   # True

# Get node class by ID
node_cls = lib.get_node_by_id("math.add")
instance = node_cls()

# Serialize library structure
lib_json = lib.full_serialize()
print(lib_json)  # {'shelves': [{'name': 'Math', 'nodes': [...], ...}]}
```

--------------------------------

### Event System for Reactive Programming in Python

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Demonstrates how to use the event system in funcnodes-core for reactive programming. It shows how to define custom nodes, attach event listeners for various node states, and trigger events.

```python
from funcnodes_core import Node, NodeInput, NodeOutput, EventEmitterMixin
from funcnodes_core.eventmanager import emit_after, emit_before

class CounterNode(Node):
    node_id = "counter"
    trigger_in = NodeInput(id="trigger", type=int, default=0)
    count_out = NodeOutput(id="count", type=int)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._count = 0

    async def func(self, trigger: int):
        self._count += 1
        self.outputs["count"].value = self._count

# Event handlers
def on_trigger_start(src, **kwargs):
    print(f"Node {src.uuid} starting trigger")

def on_trigger_done(src, **kwargs):
    print(f"Node {src.uuid} finished, result: {kwargs.get('result')}")

def on_value_set(src, result, **kwargs):
    print(f"Output value set to: {result}")

# Attach event listeners
node = CounterNode()
node.on("triggerstart", on_trigger_start)
node.on("triggerdone", on_trigger_done)
node.outputs["count"].on("after_set_value", on_value_set)

# Listen to all events with wildcard
def log_all_events(event, src, **data):
    print(f"Event: {event}")

node.on("*", log_all_events)

# Execute and observe events
node["trigger"] = 1
await node

# Remove listeners
node.off("triggerstart", on_trigger_start)
```

--------------------------------

### Instance-Bound Nodes with NodeClassMixin

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Illustrates how to create node classes bound to specific object instances using NodeClassMixin. This allows for stateful node operations where nodes can access instance-specific attributes and methods. It covers creating instances, binding nodes, and accessing node outputs.

```python
from funcnodes_core import NodeClassMixin, instance_nodefunction
import funcnodes_core as fn

class DataProcessor(NodeClassMixin):
    NODECLASSID = "data_processor"

    def __init__(self, multiplier: float):
        super().__init__()
        self.multiplier = multiplier
        self.uuid = f"processor_{id(self)}"

    @instance_nodefunction()
    def process(self, value: float) -> float:
        """Processes value with instance multiplier."""
        return value * self.multiplier

    @instance_nodefunction(trigger_on_call=True)
    def get_multiplier(self) -> float:
        """Returns the current multiplier."""
        return self.multiplier

# Create instances with different state
processor_2x = DataProcessor(multiplier=2.0)
processor_5x = DataProcessor(multiplier=5.0)

# Create nodes - each bound to its instance
processor_2x.create_nodes()
processor_5x.create_nodes()

# Get node class for this instance
NodeClass2x = processor_2x.get_nodeclass("process")
node = NodeClass2x()
node["value"] = 10.0
await node
print(node.outputs["out"].value)  # 20.0 (10 * 2)

# List all node instances for a method
all_process_nodes = processor_2x.get_nodes("process")
```

--------------------------------

### NodeSpace Management and Serialization in Python

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Shows how to create, manage, and serialize/deserialize node spaces using the NodeSpace class. It includes defining custom nodes, connecting them, setting properties, and persisting the graph state.

```python
from funcnodes_core import NodeSpace, Node, NodeInput, NodeOutput, Shelf

class ProcessorNode(Node):
    node_id = "processor"
    inp = NodeInput(id="in", type=int, default=0)
    out = NodeOutput(id="out", type=int)

    async def func(self, **kwargs):
        self.outputs["out"].value = kwargs["in"] * 2

# Create and populate nodespace
ns = NodeSpace()

# Add node shelf to library
shelf = Shelf(
    name="Processing",
    description="Data processing nodes",
    nodes=[ProcessorNode]
)
ns.add_shelf(shelf)

# Add nodes by class ID
node1 = ns.add_node_by_id("processor")
node2 = ns.add_node_by_id("processor")

# Connect nodes within the space
node1.outputs["out"].connect(node2.inputs["in"])

# Set properties
ns.set_property("workflow_name", "data_pipeline")
ns.set_secret_property("api_key", "secret123")  # Not serialized

# Serialize/deserialize the entire workspace
state = ns.serialize()
print(state)  # {'nodes': [...], 'edges': [...], 'prop': {...}, 'groups': {...}}

# Restore from serialized state
new_ns = NodeSpace()
new_ns.add_shelf(shelf)
new_ns.deserialize(state)

# Wait for all nodes to complete
await ns.await_done()
```

--------------------------------

### Dynamic Input Value Options and Constraints

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Explains how to configure input constraints and dynamically update value options for node inputs. It shows how to use `update_other_io_options` to make dropdown options dependent on other input values and how to programmatically update options and constraints for a given input.

```python
import funcnodes_core as fn
from funcnodes_core.decorator import update_other_io_options

# Dynamic dropdown options based on another input
@fn.NodeDecorator(
    node_id="data.select_key",
    default_io_options={
        "data": {
            "on": {
                "after_set_value": update_other_io_options(
                    "key",
                    lambda x: list(x.keys()) if isinstance(x, dict) else []
                )
            }
        }
    }
)
def select_key(data: dict, key: str) -> any:
    """Select a key from dictionary with dynamic dropdown."""
    return data.get(key)

node = select_key()
node["data"] = {"name": "Alice", "age": 30, "city": "NYC"}
# key input now has value_options: {"options": ["name", "age", "city"]}

node["key"] = "name"
await node
print(node.outputs["out"].value)  # "Alice"

# Manual value options with constraints
@fn.NodeDecorator(node_id="math.clamp")
def clamp_value(
    value: float,
    min_val: float = 0.0,
    max_val: float = 100.0
) -> float:
    return max(min_val, min(max_val, value))

node = clamp_value()
# Update value options programmatically
node.inputs["value"].update_value_options(min=0, max=100, step=0.1)
```

--------------------------------

### Implement Nodes Directly by Subclassing Node

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Subclassing the `Node` class provides direct control over node implementation, allowing for custom initialization, multiple I/O ports, and complex internal states. This method requires defining inputs and outputs explicitly using `NodeInput` and `NodeOutput`. Dependencies include `funcnodes_core`.

```python
import funcnodes_core as fn
from funcnodes_core import Node, NodeInput, NodeOutput

class MultiplyNode(Node):
    node_id = "math.multiply"
    node_name = "Multiply"
    description = "Multiplies two numbers"

    # Define inputs with types, defaults, and descriptions
    factor_a = NodeInput(
        id="a",
        type=float,
        default=1.0,
        description="First factor"
    )
    factor_b = NodeInput(
        id="b",
        type=float,
        default=1.0,
        description="Second factor"
    )

    # Define output
    product = NodeOutput(id="result", type=float)

    async def func(self, a: float, b: float) -> float:
        result = a * b
        self.outputs["result"].value = result
        return result

# Usage
node = MultiplyNode()
node.inputs["a"].value = 2.5
node.inputs["b"].value = 4.0
await node
print(node.outputs["result"].value)  # Output: 10.0

```

--------------------------------

### Threaded and Process Execution for Node Operations

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Shows how to run computationally intensive nodes in separate threads or processes to prevent blocking the main event loop. It demonstrates using `separate_thread=True` for I/O-bound tasks and `separate_process=True` for CPU-bound tasks, along with concurrent execution using `run_until_complete`.

```python
import funcnodes_core as fn
import time

# Separate thread execution - for I/O bound operations
@fn.NodeDecorator(node_id="io.fetch", separate_thread=True)
def fetch_data(url: str) -> dict:
    """Simulates slow I/O operation."""
    time.sleep(2)  # Simulate network delay
    return {"data": f"fetched from {url}"}

# Separate process execution - for CPU bound operations
@fn.NodeDecorator(node_id="compute.heavy", separate_process=True)
def heavy_computation(n: int) -> int:
    """CPU-intensive computation in separate process."""
    result = 0
    for i in range(n * 1000000):
        result += i
    return result

# Execute multiple nodes concurrently
fetch1 = fetch_data()
fetch2 = fetch_data()
compute = heavy_computation()

fetch1["url"] = "https://api1.example.com"
fetch2["url"] = "https://api2.example.com"
compute["n"] = 10

# All run in parallel
start = time.time()
await fn.run_until_complete(fetch1, fetch2, compute)
elapsed = time.time() - start

print(f"Completed in {elapsed:.1f}s")  # ~2s instead of 6s sequential
print(fetch1.outputs["out"].value)
print(compute.outputs["out"].value)
```

--------------------------------

### Define Connection Points with NodeInput and NodeOutput

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

NodeInput and NodeOutput classes are used to define the typed connection points (ports) on nodes. They support configuration options such as default values, required status, triggering behavior, value constraints, and enabling multiple connections. Dependencies include `funcnodes_core`.

```python
from funcnodes_core import NodeInput, NodeOutput, NoValue

# Input with validation options
input_port = NodeInput(
    id="temperature",
    name="Temperature (C)",
    type=float,
    default=20.0,
    required=True,
    does_trigger=True,  # Setting value triggers node execution
    description="Temperature in Celsius",
    value_options={"min": -273.15, "max": 1000, "step": 0.1}
)

# Optional input with NoValue default
optional_input = NodeInput(
    id="scale_factor",
    type=float,
    default=NoValue,  # Indicates no default value
    required=False
)

# Output with multiple connection support
output_port = NodeOutput(
    id="readings",
    type=list,
    allow_multiple=True  # Can connect to multiple inputs
)

# Check input readiness
print(input_port.ready())
print(input_port.value)

```

--------------------------------

### Connect Nodes to Form Workflows

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Nodes can be connected by linking their output ports to the input ports of other nodes. This establishes data flow, allowing values to propagate automatically when outputs are updated. The `connect` method is used for establishing these links. Dependencies include `funcnodes_core`.

```python
import funcnodes_core as fn

@fn.NodeDecorator(node_id="gen.constant")
def constant(value: int = 10) -> int:
    return value

@fn.NodeDecorator(node_id="math.double")
def double(x: int) -> int:
    return x * 2

@fn.NodeDecorator(node_id="math.square")
def square(x: int) -> int:
    return x ** 2

# Create node instances
source = constant()
doubler = double()
squarer = square()

# Connect nodes: source -> doubler -> squarer
source.outputs["out"].connect(doubler.inputs["x"])
doubler.outputs["out"].connect(squarer.inputs["x"])

# Execute the chain
source["value"] = 5
await fn.run_until_complete(source, doubler, squarer)

print(doubler.outputs["out"].value)  # 10
print(squarer.outputs["out"].value)  # 100

```

--------------------------------

### Custom JSON Decoding and Encoding for Complex Numbers

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Demonstrates how to register a custom decoder for complex numbers and how to serialize/deserialize data containing complex types using JSONEncoder and JSONDecoder. It also shows direct encoding of complex numbers into a specific dictionary format.

```python
from funcnodes_core.utils import JSONDecoder, JSONEncoder

# Register decoder
def complex_decoder(obj):
    if isinstance(obj, dict) and obj.get("__complex__"):
        return complex(obj["real"], obj["imag"])
    return obj

JSONDecoder.add_decoder(complex_decoder)

# Serialize/deserialize with custom types
data = {"value": complex(3, 4), "name": "test"}
encoded = json.dumps(data, cls=JSONEncoder)
decoded = json.loads(encoded, cls=JSONDecoder)
print(decoded)  # {'value': (3+4j), 'name': 'test'}

# Use apply_custom_encoding for direct encoding
result = JSONEncoder.apply_custom_encoding({"items": [1, 2, complex(1, 2)]})
print(result)  # {'items': [1, 2, {'__complex__': True, 'real': 1.0, 'imag': 2.0}]}
```

--------------------------------

### Custom JSON Serialization in Python

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Explains how to extend the JSON serialization capabilities for custom data types like complex numbers. It shows how to register custom encoders with JSONEncoder.

```python
from funcnodes_core.utils.serialization import JSONEncoder, JSONDecoder, Encdata
import json

# Register custom type encoder
def complex_encoder(obj, preview=False):
    if isinstance(obj, complex):
        return Encdata(
            data={"__complex__": True, "real": obj.real, "imag": obj.imag},
            handeled=True,
            done=True
        )
    return Encdata(data=obj, handeled=False)

JSONEncoder.add_encoder(complex_encoder, enc_cls=[complex])
```

--------------------------------

### Create Nodes from Functions using NodeDecorator

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

The `@NodeDecorator` transforms regular Python functions into Node classes, simplifying node creation with automatic input/output type inference. It allows for defining node IDs and directly using decorated functions as nodes. Dependencies include `funcnodes_core`.

```python
import funcnodes_core as fn

# Basic node creation from a function
@fn.NodeDecorator(node_id="math.add")
def add_numbers(a: int, b: int) -> int:
    """Adds two integers together."""
    return a + b

# Create node instance and execute
node = add_numbers()
node["a"] = 5
node["b"] = 3
await node  # Triggers execution
print(node.outputs["out"].value)  # Output: 8

# Multiple outputs using tuple return type
from typing import Tuple

@fn.NodeDecorator(node_id="math.divmod")
def divide_and_mod(dividend: int, divisor: int) -> Tuple[int, int]:
    """Returns quotient and remainder."""
    return dividend // divisor, dividend % divisor

node = divide_and_mod()
node["dividend"] = 17
node["divisor"] = 5
await node
# Outputs are named "out" and "out_" by default for tuples

```

--------------------------------

### Disconnect Node Output in JavaScript

Source: https://context7.com/linkdlab/funcnodes_core/llms.txt

Demonstrates how to disconnect an output of one node from the input of another in JavaScript. This is useful for managing dynamic graph structures.

```javascript
source.outputs["out"].disconnect(doubler.inputs["x"])
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.