diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml index 7e8c9cb..cab946a 100644 --- a/.github/workflows/python-publish.yml +++ b/.github/workflows/python-publish.yml @@ -36,6 +36,11 @@ jobs: with: python-version: "3.x" + - name: Sync version to sonar-project.properties + run: | + # Ensure version is synced before building + python scripts/update_version.py + - name: Build release distributions run: | # Build the hyperway package @@ -92,6 +97,11 @@ jobs: with: python-version: '3.12' + - name: Sync version to sonar-project.properties + run: | + # Ensure version is synced before SonarCloud analysis + python scripts/update_version.py + - name: Install Graphviz run: | wget https://gitlab.com/api/v4/projects/4207231/packages/generic/graphviz-releases/14.0.1/ubuntu_25.04_graphviz-14.0.1-cmake.deb -O /tmp/graphviz-14.0.1.deb diff --git a/docs/future/knuckles.md b/docs/future/knuckles.md new file mode 100644 index 0000000..76e0ff4 --- /dev/null +++ b/docs/future/knuckles.md @@ -0,0 +1,129 @@ +## Knuckles and Connection Decisions + +> A knuckle is a connection selection mechanism that filters available edges at execution time. It transforms static graph topology into dynamic routing based on runtime state, values, or conditions. + +### Core Concept + +When a `Unit` has multiple outgoing connections, the stepper normally follows all of them. A knuckle intercepts this point and returns a filtered subset: + +```python +# Pseudocode +def select_connections(connections, stepper, akw): + # connections: tuple of available Connection objects + # stepper: StepperC instance (for state access) + # akw: ArgsPack with current values + return filtered_subset_of_connections +``` + +### Implementation Approaches + +**1. Method Override** - Subclass `Unit` and override `select_connections()`: + +```python +class ConditionalUnit(Unit): + def select_connections(self, connections, stepper, akw): + if condition_met(akw): + return connections # follow all + return connections[:1] # follow only first +``` + +**2. Pluggable Knuckle** - Attach a callable to `unit.knuckle`: + +```python +unit = as_unit(func) +unit.knuckle = lambda conns, s, akw: filter_logic(conns, akw) +``` + +The `Unit.select_connections()` method checks for an attached knuckle first, then falls back to default behavior (return all connections). + +### Integration Point + +The stepper invokes knuckle logic in `call_one_unit()` after resolving available connections but before iteration: + +```python +# Pseudocode in StepperC.call_one_unit +connections = get_connections(graph, unit) +selected = unit.select_connections(connections, stepper, akw) +if not selected: + return unit.leaf(...) # treat as end-node +for conn in selected: + # execute connection... +``` + +### Use Cases + +- **Conditional branching**: Route based on value thresholds, types, or predicates +- **State machine transitions**: Implement NFA/epsilon transitions by selecting connections without consuming input +- **Round-robin distribution**: Stateful knuckles can track invocation count and alternate paths +- **Circuit breaking**: Dynamically disable connections based on errors or stepper state +- **A/B testing**: Probabilistic routing between connection sets + +### Event Emission and State Machines + +Knuckles enable independent state machine behavior: + +- A knuckle can emit events to external observers when selecting connections +- Epsilon transitions occur when `akw` values don't change but connection selection does +- Non-deterministic routing happens when multiple valid connections exist and selection varies per invocation +- State machines can track history via `stepper.stash` or custom knuckle state + +# Example + +An example of a knuckle selecting connections given their name and a key in `akw`: + +```python +def name_based_knuckle(connections, stepper, akw): + key = akw.get('route_key') + return [conn for conn in connections if conn.name == key] + +unit = as_unit(some_function) +unit.knuckle = name_based_knuckle + + +# 4 connections, two called dave: +g = Graph() + +for name in ['alice', 'bob', 'dave', 'dave']: + g.connect(unit, another_unit, name=name) +stepper = StepperC(g) + +akw1 = ArgsPack(route_key='dave') +selected_conns = unit.select_connections(g.get_connections(unit), stepper, akw1) +assert len(selected_conns) == 2 # both 'dave' connections selected +``` + +```python +akw2 = ArgsPack(route_key='alice') +selected_conns = unit.select_connections(g.get_connections(unit), stepper, akw2) +assert len(selected_conns) == 1 # only 'alice' connection selected +``` + +## Value Splitting Knuckles + +A knuckle can perform value splitting. Each connection can receive a modified `ArgsPack`: + +```python +def splitting_knuckle(connections, stepper, akw): + result = [] + for conn in connections: + modified_akw = akw.copy() + modified_akw['selected'] = conn.name + result.append((conn, modified_akw)) + return result + +unit.knuckle = splitting_knuckle +``` + +In an example where the value is numerical, the knuckle can provide each edge with a divided value: + +```python +def dividing_knuckle(connections, stepper, akw): + value = akw.get('value', 1) + num_conns = len(connections) + for conn in connections: + modified_akw = akw.copy() + modified_akw['value'] = value / num_conns + yield (conn, modified_akw) + +unit.knuckle = dividing_knuckle +``` \ No newline at end of file diff --git a/docs/future.md b/docs/future/readme.md similarity index 61% rename from docs/future.md rename to docs/future/readme.md index eaf6bdc..0312a19 100755 --- a/docs/future.md +++ b/docs/future/readme.md @@ -4,11 +4,31 @@ HyperWay interacts with state machines using non-deterministic finite automatons (NFA). This allows changing condition sets as they traverse paths or nodes. Non-deterministic functionality lets us select which steps to take going forward. This is coupled with lambda functions along the edges of nodes. When a connection is chosen or a series of connections are selected, the lambda function can serve as a state transitioning tool, allowing a parallel or sibling system to manage a state independently of the graph and its nodes. ## Knuckles and Lambda Functions + +see also: [Knuckles](./knuckles.md) + To integrate additional features into HyperWay, we extend the concept of a knuckle. A knuckle allows selecting from a finite set of connections. When visualized, this will allow using a knuckle function (or lambda) for a list of connections. The lambda function not only connects to a list of nodes, reducing the list of connections upon request, but also enables event emission for an independent state machine. This state machine can have non-deterministic or epsilon transitions, where a state can transition without consuming input symbols. ## Tagged Nodes and Edges HyperWay can utilize tagged nodes or edges in a hypergraph methodology. During a step, rather than deterministic nodes and connections, it can request or infer a new set of nodes or conditions. The request for nodes is built into the graph as a tag, enabling it to find all nodes with a particular tag. Similarly, tagged connections can be requested upon exiting a node, retrieving a list of connections defined by a tag. +## [TODO: Rename] Fiber Connections + +A _fiber_ is a "thinner" connection between nodes, or the knuckles on those nodes. A fiber can bridge the graph to reference nodes in the graph, of which are not directly connected (on the path). + +For example we have a ring of 6 nodes, but the output of node three, alters node one output value. A fiber connection can be made between these nodes, allowing node three to influence node one without a direct connection in the path. + +The _Graph_ path and nodes don't see this connection. The stepper doesn't traverse this edge. Instead, the fiber connection is a side-channel that allows nodes to influence each other without being on the main path. + +The fiber may perform a range of interactions with the destinations. The receiver node may choose to listen to the fiber connection, or ignore it. The fiber connection may also perform transformations on the data being sent, or buffer it for later use. + +--- + +Conceptually I feel this is important for the purpose of "back talk", where the tip of a connection (the outbound node or knuckle) can send information back to an earlier point in the graph without being on the main path. + +Consider for example electronic circuits, where feedback loops are common. A fiber connection could represent a feedback loop in a graph, without affecting the main flow of data. + + ## Hyperedges Hyperedges allow multiple input and exit nodes on a single connection, treating many nodes grouped under one hyperedge as a single definition. This extends to executing a sub-graph of connections that the hyperedge is connected to. Implementing a boundary around a group of nodes creates a hyperedge that can connect to another node, graph, or hyperedge. @@ -18,6 +38,47 @@ The value from a hyperedge yields the same count of results as the nodes inside ### Unique Configurations Integrating unique functionality directly into the hyperedge allows for specific manipulations, requests, or configurations unique to the edge itself. For example, a hyperedge might require all nodes on the edge to activate simultaneously or stash the results of particular nodes as the stepper walks through them. When a hyperedge is "full," it emits a result down its connections. The default method requires all nodes to be activated simultaneously, while optional methods, such as the fill method, provide flexibility in how and when results are emitted. + +## Edge | Hyperedge | Fiber | Knuckle + +An edge: + ++ Standard connection ++ Single input and output node ++ Deterministic path ++ Has Wire function ++ Does not contain a sub-graph + +A hyperedge: + ++ Multiple input and output nodes ++ Non-deterministic path ++ Can contain sub-graphs ++ Has Recombobulation function ++ multiple inputs [may] fire simultaneously for an edge reaction + +A fiber: + ++ Side-channel connection ++ Connects non-adjacent nodes ++ Independent of main graph path ++ Can influence nodes and knuckles ++ Does not affect stepper traversal ++ Does not contain a sub-graph + +A knuckle (aka connection selection): + ++ Connection selector ++ Finite set of edges ++ Can use fiber ++ Has api for connection selection + + + + + + + ### Directing a Stepper Assign directions to a stepper, applying a preferred path to follow. For example, when on node `ID1`, follow connections `{4,6}`: diff --git a/docs/future/vector-directional-execitions.md b/docs/future/vector-directional-execitions.md new file mode 100644 index 0000000..83fec7a --- /dev/null +++ b/docs/future/vector-directional-execitions.md @@ -0,0 +1,18 @@ +# Vector Directional Executions + +Vector directional executions refer to selecting nodes through directional vectors, such as PI/2. Allowing the developer to build computational graphs through directional relationships. + +> In my thoughts, I call this a "ghost edge", where we assume a (default) connection, if the directional vector plots within the radius of another node (a function with an {x,y}) + + +## Simple Example + + +A node is sitting at the origin (0,0) and has a directional vector of PI/2 (90 degrees). This means that the node is pointing straight up along the positive Y-axis. + +For a a higher selection count in a more dense space, we can consider a node at (0,0) with a radius of 5 units. Pi/2 will plot a point nearest a node at (0,5). + +This can also be combined with other directional vectors to create more complex selection patterns. For example, using PI/4 (45 degrees) and 3*PI/4 (135 degrees) will select nodes that are diagonally positioned relative to the origin node. + +Another example is an arc spread selecting nodes in an connic projection. + diff --git a/docs/sonarqube/QUICK_REFERENCE.md b/docs/sonarqube/QUICK_REFERENCE.md index 9afb854..c07de40 100644 --- a/docs/sonarqube/QUICK_REFERENCE.md +++ b/docs/sonarqube/QUICK_REFERENCE.md @@ -4,7 +4,7 @@ ```bash # Generate coverage locally -./generate_coverage.sh py312 +./scripts/generate_coverage.sh py312 # Or with tox directly tox -e py312 diff --git a/docs/sonarqube/WORKFLOW_COMPLETE.md b/docs/sonarqube/WORKFLOW_COMPLETE.md index 74b97c3..5095e8b 100644 --- a/docs/sonarqube/WORKFLOW_COMPLETE.md +++ b/docs/sonarqube/WORKFLOW_COMPLETE.md @@ -47,7 +47,7 @@ Comparison document including: - `tox.ini` - Generates coverage.xml - `sonar-project.properties` - SonarQube/SonarCloud configuration -- `generate_coverage.sh` - Helper script for local testing +- `scripts/generate_coverage.sh` - Helper script for local testing --- @@ -216,10 +216,10 @@ Current Project Coverage (as of last run): The workflows integrate seamlessly with: ``` -tox.ini # Already generates coverage.xml -sonar-project.properties # Already configured -generate_coverage.sh # Can test locally -pyproject.toml # Tox dependency already added +tox.ini # Already generates coverage.xml +sonar-project.properties # Already configured +scripts/generate_coverage.sh # Can test locally +pyproject.toml # Tox dependency already added ``` ### No Changes Needed To @@ -282,7 +282,7 @@ on: #### 1. Coverage.xml Not Found ```bash # Verify locally -./generate_coverage.sh py312 +./scripts/generate_coverage.sh py312 ls -lh coverage.xml ``` @@ -360,11 +360,13 @@ Add to `README.md`: └── WORKFLOW_COMPLETE.md # This file Project Root: -├── tox.ini # Tox configuration (updated) -├── sonar-project.properties # Sonar configuration (updated) -├── generate_coverage.sh # Coverage helper script -├── SONARQUBE_SETUP.md # Original setup doc (updated) -└── coverage.xml # Generated by tox (not committed) +├── scripts/ +│ ├── generate_coverage.sh # Coverage helper script +│ └── update_version.py # Version sync script +├── tox.ini # Tox configuration (updated) +├── sonar-project.properties # Sonar configuration (updated) +├── SONARQUBE_SETUP.md # Original setup doc (updated) +└── coverage.xml # Generated by tox (not committed) ``` --- @@ -391,7 +393,7 @@ Project Root: 6. ✅ Comparison guide for choosing platform 7. ✅ Troubleshooting guide included 8. ✅ Badge examples provided -9. ✅ Local testing verified with generate_coverage.sh +9. ✅ Local testing verified with scripts/generate_coverage.sh 10. ✅ Works with existing project structure **Ready to deploy!** Choose your platform and follow the Quick Start Guide above. diff --git a/docs/version-management.md b/docs/version-management.md new file mode 100644 index 0000000..4c4c39f --- /dev/null +++ b/docs/version-management.md @@ -0,0 +1,73 @@ +# Version Management + +The version is defined in **`src/hyperway/_version.py`**: + +```python +__version__ = "1.0.5" +``` + +The `pyproject.toml` uses dynamic versioning to read from the `_version.py` file: + +```toml +[project] +dynamic = ["version"] + +[tool.setuptools.dynamic] +version = {attr = "hyperway._version.__version__"} +``` + +`sonar-project.properties` is a plain text config file, we use a Python script to sync it: + +```bash +python scripts/update_version.py +``` + +This script: +1. Reads the version from `src/hyperway/_version.py` +2. Updates the `sonar.projectVersion` line in `sonar-project.properties` + +## Updating the Version + +**To update the version for a new release:** + +1. Edit **only** the version in `src/hyperway/_version.py`: + ```python + __version__ = "1.0.6" + ``` + +2. (Optional) Run the update script to sync SonarQube config: + ```bash + python scripts/update_version.py + ``` + > **Note:** The GitHub Actions workflow automatically runs this script during release builds, so you don't have to remember to do it manually. + +3. Commit the version file: + ```bash + git add src/hyperway/_version.py + # Optionally include sonar-project.properties if you ran scripts/update_version.py + git commit -m "Bump version to 1.0.6" + ``` + +### Automated Workflow + +The GitHub Actions workflow (`.github/workflows/python-publish.yml`) automatically syncs the version before: +- **Building releases** - Ensures the package has the correct version +- **SonarCloud analysis** - Ensures the analysis reports the correct version + +This means even if you forget to run `scripts/update_version.py` locally, the CI/CD pipeline will ensure everything stays in sync. + +## Verification + +Check that the version is correctly set: + +```bash +# Check package version +python -c "import sys; sys.path.insert(0, 'src'); from hyperway import __version__; print(__version__)" + +# Check SonarQube version +grep "sonar.projectVersion" sonar-project.properties + +# Or run the sync script to verify +python scripts/update_version.py +``` + diff --git a/pyproject.toml b/pyproject.toml index 526c4ac..f2658ee 100755 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "hyperway" -version = "1.0.5" +dynamic = ["version"] authors = [ { name="Strangemother", email="hyperway@strangemother.com" }, ] @@ -61,6 +61,9 @@ docs = [ package-dir = {"" = "src"} packages = ["hyperway", "hyperway.graph"] +[tool.setuptools.dynamic] +version = {attr = "hyperway._version.__version__"} + [tool.setuptools.package-data] hyperway = ["**/*.html", "**/*.js", "**/*.css"] diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 0000000..3d9e6f7 --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,57 @@ +# Build Scripts + +This directory contains build and release automation scripts for the python-hyperway project. + +## Scripts + +### `update_version.py` + +Synchronizes the version number from `src/hyperway/_version.py` to `sonar-project.properties`. + +**Usage:** +```bash +python scripts/update_version.py +``` + +**What it does:** +1. Imports the current version from `hyperway.__version__` +2. Updates the `sonar.projectVersion` line in `sonar-project.properties` + +**Automated Usage:** +This script is automatically run by GitHub Actions during: +- Release builds (before building the package) +- SonarCloud analysis (before running analysis) + +See [Version Management Documentation](../docs/version-management.md) for more details. + +--- + +### `generate_coverage.sh` + +Generates code coverage reports for SonarQube/SonarCloud analysis. + +**Usage:** +```bash +# Run all configured Python versions +./scripts/generate_coverage.sh + +# Run specific Python version +./scripts/generate_coverage.sh py312 +``` + +**What it does:** +1. Runs tests using `tox` with coverage enabled +2. Generates `coverage.xml` in the project root +3. Provides feedback on coverage report generation +4. Verifies the coverage file was created successfully + +**Output:** +- `coverage.xml` - Code coverage report in Cobertura format + +See [SonarQube Documentation](../docs/sonarqube/) for more details. + +--- + +## Future Scripts + +Additional build and release automation tools will be added to this directory as needed. diff --git a/generate_coverage.sh b/scripts/generate_coverage.sh similarity index 87% rename from generate_coverage.sh rename to scripts/generate_coverage.sh index 98649ab..01edf6f 100755 --- a/generate_coverage.sh +++ b/scripts/generate_coverage.sh @@ -1,10 +1,10 @@ #!/bin/bash # Generate coverage report for SonarQube -# Usage: ./generate_coverage.sh [tox-environment] +# Usage: ./scripts/generate_coverage.sh [tox-environment] # # Examples: -# ./generate_coverage.sh # Run all available Python versions -# ./generate_coverage.sh py312 # Run only Python 3.12 +# ./scripts/generate_coverage.sh # Run all available Python versions +# ./scripts/generate_coverage.sh py312 # Run only Python 3.12 set -e diff --git a/scripts/update_version.py b/scripts/update_version.py new file mode 100755 index 0000000..9815445 --- /dev/null +++ b/scripts/update_version.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +"""Update version in sonar-project.properties from package version.""" + +import re +import sys +from pathlib import Path + +# Get the project root (parent of scripts directory) +PROJECT_ROOT = Path(__file__).parent.parent + +# Add src to path to import hyperway +sys.path.insert(0, str(PROJECT_ROOT / "src")) + +from hyperway import __version__ + + +def get_package_version(): + """Read version from hyperway package.""" + return __version__ + + +def update_sonar_properties(version): + """Update version in sonar-project.properties.""" + sonar_file = PROJECT_ROOT / "sonar-project.properties" + + with open(sonar_file) as f: + content = f.read() + + # Replace the version line + new_content = re.sub( + r'sonar\.projectVersion=.*', + f'sonar.projectVersion={version}', + content + ) + + with open(sonar_file, 'w') as f: + f.write(new_content) + + print(f"✓ Updated sonar-project.properties to version {version}") + + +if __name__ == "__main__": + version = get_package_version() + print(f"Package version: {version}") + update_sonar_properties(version) diff --git a/sonar-project.properties b/sonar-project.properties index 7330dfa..54f9ab1 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -5,6 +5,7 @@ sonar.projectKey=Strangemother_python-hyperway sonar.organization=strangemother sonar.projectName=Python Hyperway +## note: Managed automatically via scripts/update_version.py sonar.projectVersion=1.0.5 # Source code settings diff --git a/src/hyperway/__init__.py b/src/hyperway/__init__.py index 9390a3f..0def203 100755 --- a/src/hyperway/__init__.py +++ b/src/hyperway/__init__.py @@ -1,3 +1,4 @@ +from ._version import __version__ from . import edges from . import nodes @@ -7,4 +8,5 @@ from .edges import make_edge, Connection from .graph import Graph from .packer import argspack, argpack, ArgsPack -from .stepper import StepperException \ No newline at end of file +from .stepper import StepperException +from .constants import INITIATE_DISTRIBUTED, INITIATE_UNIFIED \ No newline at end of file diff --git a/src/hyperway/_version.py b/src/hyperway/_version.py new file mode 100644 index 0000000..e171cd6 --- /dev/null +++ b/src/hyperway/_version.py @@ -0,0 +1,2 @@ +"""Version information for hyperway.""" +__version__ = "1.0.5" diff --git a/src/hyperway/constants.py b/src/hyperway/constants.py new file mode 100644 index 0000000..7b65d9f --- /dev/null +++ b/src/hyperway/constants.py @@ -0,0 +1,9 @@ +"""Constants for the hyperway package. + +This module contains package-wide constants to avoid magic strings +and prevent circular import issues. +""" + +# Stepper initiation mode constants +INITIATE_DISTRIBUTED = 'distributed' # Edge-centric: call node once per connection +INITIATE_UNIFIED = 'unified' # Node-centric: call node once, fan out to connections diff --git a/src/hyperway/edges.py b/src/hyperway/edges.py index 04ca630..1d24f2e 100755 --- a/src/hyperway/edges.py +++ b/src/hyperway/edges.py @@ -84,15 +84,23 @@ def as_connections(*items, graph=None): return res -def get_connections(graph, unit): +def get_connections(graph, unit, akw=None): + """ + Given a graph and a unit (node or connection), return the connections from that unit. + This function checks if the unit has a custom get_connections() method; + if so, it uses that method to retrieve the connections. Otherwise, it looks up + the connections in the graph using the unit's ID. + + The optional `akw` is pushed through to the unit.get_connections() if it exists. + """ # resolve the list of connections using the # id of the unit. # print('xx - Get for', unit) if hasattr(unit, 'get_connections'): # Is an edge. print('Using unit.get_connections') - res = unit.get_connections(graph) + res = unit.get_connections(graph, akw=akw) else: res = graph.get(as_unit(unit).id(), None) @@ -266,7 +274,7 @@ def merge_node(self): def b(self): return self.parent_connection.get_b() - def get_connections(self, graph): + def get_connections(self, graph, akw=None): """Return the connections from node B, as the next step. This is called by the stepper when processing this partial connection.""" resolve = graph.resolve_node_connections diff --git a/src/hyperway/graph/graph.py b/src/hyperway/graph/graph.py index 2226b59..5a66292 100644 --- a/src/hyperway/graph/graph.py +++ b/src/hyperway/graph/graph.py @@ -3,6 +3,7 @@ from .base import GraphBase, connect, add, resolve from .. import writer from ..packer import argspack +from ..constants import INITIATE_DISTRIBUTED __all__ = ['Graph'] @@ -12,6 +13,11 @@ class Graph(GraphBase): _stepper_args = None _stepper_rows = None _stepper_class = None + _stepper_initiate = None + + # In Graph theory, the 'seed' is the primary point of initiation. + # Here, we use 'initiate' as the key applied within the prepare + seed_flag_parameter = 'initiate' # Parameter name for initiate mode in stepper def get_stepper_class(self): if self._stepper_class is None: @@ -97,6 +103,7 @@ def resolve_node(self, node): def stepper_prepare(self, n=None, *a, **kw): self._stepper_callers = n self._stepper_args = argspack(*a, **kw) + self._stepper_initiate = kw.pop(self.seed_flag_parameter, INITIATE_DISTRIBUTED) def stepper_prepare_many(self, *rows): """Given many rows of `node, primivites`, @@ -118,11 +125,12 @@ def stepper(self, n=None, *a, **kw): stepper = _stepper_class(self) n = n or self._stepper_callers akw = self._stepper_args + initiate_mode = self._stepper_initiate # Default: INITIATE_DISTRIBUTED if len(a) + len(kw) > 0: akw = argspack(*a, **kw) if n is not None: - stepper.prepare(n, akw=akw) + stepper.prepare(n, akw=akw, initiate=initiate_mode) return stepper def write(self, *a, **kw): diff --git a/src/hyperway/packer.py b/src/hyperway/packer.py index 8e5975b..99c4799 100644 --- a/src/hyperway/packer.py +++ b/src/hyperway/packer.py @@ -94,6 +94,9 @@ def a(self): @property def kw(self): return self.kwargs + + def get(self, key, default=None): + return self.kwargs.get(key, default) def __str__(self): return self.as_str() diff --git a/src/hyperway/stepper.py b/src/hyperway/stepper.py index 1996ca0..72d4d88 100755 --- a/src/hyperway/stepper.py +++ b/src/hyperway/stepper.py @@ -6,6 +6,7 @@ from .edges import (Connection, as_connections, is_edge, PartialConnection, get_connections) from .graph.base import is_graph +from .constants import INITIATE_DISTRIBUTED, INITIATE_UNIFIED class StepperException(Exception): @@ -100,6 +101,68 @@ def set_global_expand(expand_func): expand = expand_func +def expand_distributed(stepper, start_nodes, start_akw): + """Expand for distributed initiation: call each start node once per connection. + + This is the standard edge-centric expansion mode where each outgoing + connection from a start node results in a separate call to that node. + + Args: + stepper: The StepperC instance + start_nodes: Tuple of start node(s) + start_akw: Initial argument pack + Returns: + + Tuple of (Connection, argspack) pairs ready for the next step + """ + return expand(start_nodes, start_akw) + +def expand_unified(stepper, start_nodes, akw): + """Expand for unified initiation: call each start node once, then fan out to connections. + + This is an alternative to the standard expand() that implements node-centric + initialization. Instead of calling the node once per connection, this calls + the node once and distributes the result to all outgoing connections. + + Args: + stepper: The StepperC instance + start_nodes: Tuple of start node(s) + akw: Initial argument pack + + Returns: + Tuple of (PartialConnection, argspack) pairs ready for the next step + """ + all_rows = () + + for node in start_nodes: + # Get connections for this node + conns = get_connections(stepper.graph, node, akw=akw) + + if conns is None: + # No connections - handle as leaf + all_rows += node.leaf(stepper, akw) + continue + + # Call node ONCE + if is_unit(node): + result = node.process(*akw.a, **akw.kw) + else: + # For raw callables + result = node(*akw.a, **akw.kw) + + result_akw = argspack(result) + + # Create rows for each connection using the shared result + # Each connection's wire function will receive the same result + for conn in conns: + # Create a PartialConnection (skipping the A call since we already did it) + # The PartialConnection represents the [wire]->B portion + partial = PartialConnection(conn) + all_rows += ((partial, result_akw),) + + return all_rows + + def stepper_c(graph, start_node, argspack): stepper = StepperC(graph) res = stepper.start(start_node, akw=argspack) @@ -108,7 +171,20 @@ def stepper_c(graph, start_node, argspack): class StepperIterator(object): - + """Iterator wrapper for StepperC that enables Python iteration protocol. + + Yields successive row sets from the stepper until the graph execution + completes (rows become empty). Each yielded value is a tuple of + (next_caller, argspack) pairs representing the current execution frontier. + + Example: + >>> g = Graph() + >>> g.connect(f.add_1, f.add_2, f.add_3) + >>> s = g.stepper() + >>> s.prepare(f.add_1, akw=argspack(10)) + >>> for rows in s.iterator(): + ... print(len(rows)) # prints row count per step + """ def __init__(self, stepper, funcs, akw, **config): self.stepper = stepper self.start_nodes = funcs @@ -140,13 +216,16 @@ def is_merge_node(next_caller): class StepperC(object): """This stepper will work with functions - or just callers, and argpacks """ - concat_aware = False + # When True, enables row_concat() to merge multiple incoming rows targeting the same merge_node + concat_aware = False + # When True, stores branch-end results in stash; when False, returns rows with None as next caller + stash_ends = True + initiate = INITIATE_DISTRIBUTED # INITIATE_DISTRIBUTED (default) or INITIATE_UNIFIED def __init__(self, graph, rows=None): self.graph = graph self.run = 1 - self.stash_ends = True self.reset_stash() self.start_nodes = None @@ -156,12 +235,20 @@ def __init__(self, graph, rows=None): def reset_stash(self): self.stash = defaultdict(tuple) - def prepare(self, *funcs, akw): + def prepare(self, *funcs, akw, initiate=INITIATE_DISTRIBUTED): """Prepare the stepper with the start nodes and the initial argument pack. Next iterations will yield steps. + + Args: + *funcs: Start node(s) for execution + akw: Initial argument pack + initiate: Initial execution mode + INITIATE_DISTRIBUTED (default) - Call start node once per connection (edge-centric) + INITIATE_UNIFIED - Call start node once, then distribute result to all connections """ self.start_nodes = funcs self.start_akw = akw + self.initiate = initiate def __iter__(self): """Call upon an iterator to yield the stepper per next() interaction: @@ -199,7 +286,15 @@ def step(self, rows=None, count=1): if st_nodes is None: # Start node must be something... raise StepperException('start_nodes is None') - self.rows = rows or self.rows or expand(st_nodes, self.start_akw,) + + # Initialize rows if needed - check initiate mode for first step + if rows is None and self.rows is None: + # First step - use appropriate expansion based on initiate mode + func = expand_unified if self.initiate == INITIATE_UNIFIED else expand_distributed + # Default INITIATE_DISTRIBUTED mode - standard edge-centric expansion + self.rows = func(self, st_nodes, self.start_akw) + else: + self.rows = rows or self.rows while c < count: c += 1 @@ -256,7 +351,7 @@ def call_rows(self, rows): res += add_rows return res """ - + if self.concat_aware: rows = self.row_concat(rows) @@ -393,7 +488,7 @@ def call_one_connection(self, edge, akw): call. If a function, the _result_ is pushed into the future call stack. """ - a_to_b_conns = get_connections(self.graph, edge) + a_to_b_conns = get_connections(self.graph, edge, akw=akw) raw_res = edge.stepper_call(akw, stepper=self) res_akw = argspack(raw_res) @@ -408,7 +503,7 @@ def call_one_callable(self, func, akw): call. If a function, the _result_ is pushed into the future call stack. """ - a_to_b_conns = get_connections(self.graph, func) + a_to_b_conns = get_connections(self.graph, func, akw=akw) raw_res = func(*akw.a,**akw.kw) res_akw = argspack(raw_res) @@ -422,7 +517,7 @@ def call_one_partial_connection(self, partial_conn, akw): returning the B node raw result. """ wire_raw_res = partial_conn.stepper_call(akw, stepper=self) - b_conns = get_connections(self.graph, partial_conn.b) + b_conns = get_connections(self.graph, partial_conn.b, akw=akw) # The raw wire result here, is the wire -> B result (as the # Therefore collect the B node connections(.A), for the next calls @@ -444,7 +539,7 @@ def call_one_unit(self, unit, akw): the connection; [wire] -> B """ # where unit == a - a_to_b_conns = get_connections(self.graph, unit) + a_to_b_conns = get_connections(self.graph, unit, akw=akw) if a_to_b_conns is None: # This node call has no connection, assume an end; diff --git a/tests/test_edges.py b/tests/test_edges.py index eb22fdb..8b0ada2 100644 --- a/tests/test_edges.py +++ b/tests/test_edges.py @@ -782,6 +782,87 @@ def test_get_connections_with_edge_returns_next_nodes(self): # The result should contain func_b (the A node of edge2) self.assertEqual(len(connections), 1) self.assertEqual(connections[0], edge2.a) + + def test_get_connections_passes_akw_to_unit_method(self): + """get_connections passes akw parameter to unit.get_connections when available. + + When a unit has a custom get_connections method, and akw is provided, + the get_connections function should pass akw through to the unit's method. + This enables data-driven edge selection (knuckles functionality). + """ + from unittest.mock import MagicMock + + g = Graph() + + # Create a unit with a mock get_connections method + unit_a = as_unit(func_a) + mock_get_connections = MagicMock(return_value=()) + unit_a.get_connections = mock_get_connections + + # Create some edges + g.add(unit_a, func_b) + + # Call get_connections with akw + test_akw = argspack('test_data', value=42) + with patch('builtins.print'): # Suppress debug output + get_connections(g, unit_a, akw=test_akw) + + # Verify the unit's get_connections was called with graph and akw as keyword arg + mock_get_connections.assert_called_once_with(g, akw=test_akw) + + def test_get_connections_without_akw_parameter(self): + """get_connections works without akw parameter (backward compatibility). + + Verifies that when akw is not provided, the function still works + as before. This ensures backward compatibility with existing code. + """ + from hyperway.nodes import Unit + + g = Graph() + + # Create a unit with get_connections that handles None akw + class SafeUnit(Unit): + def get_connections(self, graph, akw=None): + """Return all connections, with optional akw.""" + return tuple(graph.get(self.id())) + + unit_a = SafeUnit(func_a) + edge1 = g.add(unit_a, func_b) + edge2 = g.add(unit_a, func_c) + + # Call without akw parameter + with patch('builtins.print'): + connections = get_connections(g, unit_a) + + # Should return all connections + self.assertEqual(len(connections), 2) + self.assertIn(edge1, connections) + self.assertIn(edge2, connections) + + def test_get_connections_with_none_akw(self): + """get_connections handles akw=None explicitly passed. + + Tests that explicitly passing akw=None works correctly and + is distinguishable from not passing akw at all (though they + should behave the same). + """ + from unittest.mock import MagicMock + + g = Graph() + + # Create a unit with a mock get_connections method + unit_a = as_unit(func_a) + mock_get_connections = MagicMock(return_value=()) + unit_a.get_connections = mock_get_connections + + g.add(unit_a, func_b) + + # Call with explicit akw=None + with patch('builtins.print'): + get_connections(g, unit_a, akw=None) + + # Verify called with None as keyword arg + mock_get_connections.assert_called_once_with(g, akw=None) class TestGetConnectionsIfBranch(unittest.TestCase): """Test get_connections if branch (line 42) - PartialConnection with get_connections method.""" diff --git a/tests/test_knuckles.py b/tests/test_knuckles.py new file mode 100644 index 0000000..64a6891 --- /dev/null +++ b/tests/test_knuckles.py @@ -0,0 +1,156 @@ +""" +Tests for the knuckles pattern - custom edge selection via unit.get_connections(). + +The knuckles pattern allows nodes to dynamically control which edges are traversed +based on runtime data (akw). This is implemented by defining a custom get_connections() +method on a Unit subclass that filters connections based on the akw parameter. +""" + +import unittest +from unittest.mock import patch + +from hyperway.edges import get_connections +from hyperway.nodes import Unit, as_unit +from hyperway.packer import argspack +from hyperway.graph import Graph + +from tiny_tools import passthrough + + +# Test functions +func_a = passthrough +func_b = passthrough + +def func_c(v=None): + """Standard test function C.""" + if v is None: + v = 0 + return v + 3 + + +class TestKnucklesPattern(unittest.TestCase): + """Test the knuckles pattern - data-driven edge selection.""" + + def test_custom_get_connections_filters_by_name(self): + """Unit with custom get_connections can filter edges by connection name. + + This is the core knuckles pattern: a node decides which edges to traverse + based on data in the akw parameter. Here we filter by connection name. + """ + + g = Graph() + + # Create a custom unit that filters connections by name + class SelectiveUnit(Unit): + def get_connections(self, graph, akw=None): + """Return connections filtered by akw data.""" + all_connections = tuple(graph.get(self.id())) + + if akw is None or not akw.args: + return all_connections + + # Filter by target name in akw + target = akw.args[0] if akw.args else None + filtered = tuple(c for c in all_connections if c.name == target) + return filtered if filtered else all_connections + + # Create unit and named connections + unit_a = SelectiveUnit(func_a) + edge1 = g.add(unit_a, func_b) + edge2 = g.add(unit_a, func_c) + + edge1.name = 'path_b' + edge2.name = 'path_c' + + # Test filtering to path_b + akw_b = argspack('path_b') + with patch('builtins.print'): + connections = get_connections(g, unit_a, akw=akw_b) + + self.assertEqual(len(connections), 1) + self.assertIn(edge1, connections) + self.assertNotIn(edge2, connections) + + # Test filtering to path_c + akw_c = argspack('path_c') + with patch('builtins.print'): + connections = get_connections(g, unit_a, akw=akw_c) + + self.assertEqual(len(connections), 1) + self.assertIn(edge2, connections) + self.assertNotIn(edge1, connections) + + def test_custom_get_connections_returns_all_when_no_akw(self): + """Custom get_connections falls back to all edges when akw is None.""" + + g = Graph() + + class SelectiveUnit(Unit): + def get_connections(self, graph, akw=None): + all_connections = tuple(graph.get(self.id())) + + if akw is None or not akw.args: + return all_connections + + target = akw.args[0] if akw.args else None + filtered = tuple(c for c in all_connections if c.name == target) + return filtered if filtered else all_connections + + unit_a = SelectiveUnit(func_a) + edge1 = g.add(unit_a, func_b) + edge2 = g.add(unit_a, func_c) + + edge1.name = 'path_b' + edge2.name = 'path_c' + + # Without akw, should return all connections + with patch('builtins.print'): + connections = get_connections(g, unit_a) + + self.assertEqual(len(connections), 2) + self.assertIn(edge1, connections) + self.assertIn(edge2, connections) + + def test_custom_get_connections_with_dict_routing(self): + """Knuckles pattern with dictionary-based routing data. + + Demonstrates using a dictionary in akw to make routing decisions, + showing flexibility beyond simple string matching. + """ + + g = Graph() + + class DictRoutingUnit(Unit): + def get_connections(self, graph, akw=None): + all_connections = tuple(graph.get(self.id())) + + if akw is None or not akw.kwargs: + return all_connections + + # Route based on 'target' key in kwargs + target = akw.kwargs.get('target') + if target is None: + return all_connections + + filtered = tuple(c for c in all_connections if c.name == target) + return filtered if filtered else all_connections + + unit_a = DictRoutingUnit(func_a) + edge1 = g.add(unit_a, func_b) + edge2 = g.add(unit_a, func_c) + + edge1.name = 'option_1' + edge2.name = 'option_2' + + # Route using kwargs + akw = argspack(target='option_1') + with patch('builtins.print'): + connections = get_connections(g, unit_a, akw=akw) + + self.assertEqual(len(connections), 1) + self.assertIn(edge1, connections) + self.assertNotIn(edge2, connections) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_packer.py b/tests/test_packer.py index 17feb9b..03c356e 100644 --- a/tests/test_packer.py +++ b/tests/test_packer.py @@ -167,6 +167,30 @@ def test_kw_property_shorthand(self): self.assertEqual(akw.kw, {'x': 5, 'y': 10}) +class TestArgsPackGet(unittest.TestCase): + """Test ArgsPack.get() method for accessing kwargs.""" + + def test_get_existing_key(self): + """get() returns value for existing key.""" + akw = ArgsPack(foo='bar', count=42) + + self.assertEqual(akw.get('foo'), 'bar') + self.assertEqual(akw.get('count'), 42) + + def test_get_missing_key_returns_none(self): + """get() returns None for missing key by default.""" + akw = ArgsPack(foo='bar') + + self.assertIsNone(akw.get('missing')) + + def test_get_with_default(self): + """get() returns custom default for missing key.""" + akw = ArgsPack(foo='bar') + + self.assertEqual(akw.get('missing', 'default_value'), 'default_value') + self.assertEqual(akw.get('missing', 0), 0) + + class TestArgsPackStringRepresentation(unittest.TestCase): """Test string representation methods.""" diff --git a/tests/test_stepper_advanced.py b/tests/test_stepper_advanced.py index c88747d..af060d7 100644 --- a/tests/test_stepper_advanced.py +++ b/tests/test_stepper_advanced.py @@ -31,6 +31,11 @@ is_merge_node, expand, ) +# from hyperway import stepper +from hyperway.packer import ArgsPack +from hyperway.stepper import expand_unified +from unittest.mock import Mock, patch +from hyperway.edges import Connection def multiply_by_2(v): @@ -130,6 +135,126 @@ def test_set_global_expand(self): set_global_expand(original_expand) +class TestExpandUnified(unittest.TestCase): + """Test expand_unified function for unified initiation mode.""" + + def test_expand_unified_calls_node_once(self): + """expand_unified calls each node once and distributes result to connections.""" + g = Graph() + + # Track how many times the node is called + call_count = [0] + + def counting_func(x): + call_count[0] += 1 + return x * 2 + + unit_a = as_unit(counting_func) + g.add(unit_a, multiply_by_2) + g.add(unit_a, multiply_by_2) # Two connections from unit_a + + stepper = StepperC(g) + akw = argspack(10) + + rows = expand_unified(stepper, (unit_a,), akw) + + # Node should be called once, not twice + self.assertEqual(call_count[0], 1) + # Should return 2 rows (one per connection) + self.assertEqual(len(rows), 2) + + def test_expand_unified_returns_partial_connections(self): + """expand_unified returns PartialConnection instances.""" + g = Graph() + unit_a = as_unit(multiply_by_2) + g.add(unit_a, multiply_by_2) + + stepper = StepperC(g) + akw = argspack(5) + + rows = expand_unified(stepper, (unit_a,), akw) + + self.assertEqual(len(rows), 1) + next_caller, result_akw = rows[0] + self.assertIsInstance(next_caller, PartialConnection) + # Result should be from calling the node once + self.assertEqual(result_akw.args[0], 10) + + def test_expand_unified_handles_no_connections(self): + """expand_unified handles leaf nodes with no connections.""" + from hyperway.stepper import expand_unified + + g = Graph() + unit_a = as_unit(multiply_by_2) + # No connections added - unit_a is a leaf + + stepper = StepperC(g) + akw = argspack(5) + + rows = expand_unified(stepper, (unit_a,), akw) + + # Leaf nodes return empty tuple by default + self.assertEqual(rows, ()) + + +class TestExpandUnifiedNodeCall(unittest.TestCase): + """Test expand_unified if/else for is_unit check - lines 128-135 in stepper.py.""" + + def test_is_unit_true_calls_process(self): + """When is_unit(node) is True, calls node.process().""" + g = Graph() + unit_a = as_unit(multiply_by_2) + g.add(unit_a, multiply_by_2) + + stepper = StepperC(g) + akw = argspack(5) + + rows = expand_unified(stepper, (unit_a,), akw) + + # Unit.process called: 5 * 2 = 10 + _, result_akw = rows[0] + self.assertEqual(result_akw.args[0], 10) + + def test_is_unit_false_calls_directly(self): + """When is_unit(node) is False, calls node directly (else branch).""" + g = Graph() + # Mock callable that is NOT a Unit + mock_callable = Mock(return_value=42) + mock_callable.__name__ = 'mock_func' + + unit_b = as_unit(multiply_by_2) + + # Manually create connection + conn = Connection(mock_callable, unit_b) + g.add_edge(conn) + + stepper = StepperC(g) + akw = argspack(10, foo='bar') + + # Mock get_connections to return proper list + with patch('hyperway.stepper.get_connections', return_value=[conn]): + rows = expand_unified(stepper, (mock_callable,), akw) + + # Should call mock_callable directly (else branch: not .process) + mock_callable.assert_called_once_with(10, foo='bar') + _, result_akw = rows[0] + self.assertEqual(result_akw.args[0], 42) + + def test_both_branches_return_argspack(self): + """Both if and else branches wrap result in argspack.""" + g = Graph() + unit_a = as_unit(multiply_by_2) + g.add(unit_a, multiply_by_2) + + stepper = StepperC(g) + akw = argspack(5) + + rows = expand_unified(stepper, (unit_a,), akw) + + _, result_akw = rows[0] + self.assertIsInstance(result_akw, ArgsPack) + + class TestHelperFunctions(unittest.TestCase): """Test top-level helper functions.""" diff --git a/workspace/edge-selection-v1.py b/workspace/edge-selection-v1.py new file mode 100644 index 0000000..708ad57 --- /dev/null +++ b/workspace/edge-selection-v1.py @@ -0,0 +1,177 @@ +""" +Edge Selection Example - Custom get_connections() + +This example demonstrates how a Unit can implement custom connection selection +by overriding get_connections(). + +Node "Lana" connects to four nodes through named connections: +- Connection to Bob (name='bob') +- Connection to Alice (name='alice') +- Connection to Dave1 (name='Dave') +- Connection to Dave2 (name='Dave') + +The SelectiveUnit reads the 'to' field from the data dict during process(), +then uses that value in get_connections() to filter which connections to follow. + +When data {"foo": "Bar", "to": "Dave"} flows through: +1. Lana's process() extracts to="Dave" +2. Lana's get_connections() returns only connections named "Dave" +3. Both Dave1 and Dave2 are called with the same dict +4. Bob and Alice are skipped entirely + +Result: Dynamic routing based on data flowing through the graph! +""" + +from hyperway.graph import Graph +from hyperway.nodes import Unit, as_unit +from hyperway.packer import argspack +from hyperway.stepper import StepperC + + +def print_node(name): + """Factory function to create a print function with a fixed name.""" + def printer(*args, **kwargs): + if args and isinstance(args[0], dict): + print(f" {name} received dict: {args[0]}") + else: + print(f" {name} called with: args={args}, kwargs={kwargs}") + printer.__name__ = name + return printer + + +class SelectiveUnit(Unit): + """A Unit that selects connections based on runtime data. + + Stores routing dict and overrides get_connections() to filter by name. + The dict must be set before stepper execution via set_routing_data(). + """ + + def __init__(self, func, **node_kwargs): + super().__init__(func, **node_kwargs) + self._routing_data = {} + + def set_routing_data(self, data): + """Set the routing data dict before execution.""" + self._routing_data = data + print(f"\n{self.get_name()} configured with routing data: {data}") + + def process(self, *a, **kw): + """Process and pass through the data.""" + print(f" {self.get_name()} processing data: {self._routing_data}") + + # Call the underlying function with the routing data + super().process(self._routing_data) + + # Return the data to pass it along + return self._routing_data + + def get_connections(self, graph): + """Filter connections by name based on routing data. + + This method is called by get_connections() in edges.py. + It uses the _routing_data dict to determine which connections to follow. + """ + # Get all connections from graph + connections = graph.get(self.id(), None) + + if not connections: + return None + + target_name = self._routing_data.get('to', None) + + print(f"\n{self.get_name()} selecting from {len(connections)} connections:") + print(f" Looking for connections named: '{target_name}'") + + if target_name is None: + print(" No 'to' field in routing data, returning all connections") + return connections + + # Filter connections by name + filtered = tuple(c for c in connections if c.name == target_name) + + print(f" Found {len(filtered)} matching connection(s):") + for conn in filtered: + print(f" - Connection to {conn.b.get_name()} (name='{conn.name}')") + + return filtered if filtered else None + + +def main(): + """Build and execute the example graph.""" + + # Create the graph + g = Graph() + + # Create nodes - Lana uses SelectiveUnit for custom connection selection + lana = SelectiveUnit(print_node("Lana")) + bob_node = as_unit(print_node("Bob")) + alice_node = as_unit(print_node("Alice")) + dave1_node = as_unit(print_node("Dave1")) + dave2_node = as_unit(print_node("Dave2")) + + # Configure routing data BEFORE adding connections + start_data = {"foo": "Bar", "to": "Dave"} + lana.set_routing_data(start_data) + + # Connect Lana to multiple nodes with named connections + g.add(lana, bob_node, name="bob") + g.add(lana, alice_node, name="alice") + g.add(lana, dave1_node, name="Dave") + g.add(lana, dave2_node, name="Dave") + + print("=" * 70) + print("Graph Structure:") + print("=" * 70) + print(f"Lana connects to:") + print(f" - Bob (name='bob')") + print(f" - Alice (name='alice')") + print(f" - Dave1 (name='Dave')") + print(f" - Dave2 (name='Dave')") + print() + + # Prepare and run the stepper + print("=" * 70) + print("Starting Stepper") + print("=" * 70) + + # Prepare stepper (data already set on lana node) + g.stepper_prepare(lana) + stepper = g.stepper() + + print(f"\nStarting at Lana") + print() + + # Execute the graph + step_count = 0 + while True: + step_count += 1 + print(f"\n--- Step {step_count} ---") + rows = stepper.step() + + if not rows: + print("No more rows to process") + break + + print(f"Active rows: {len(rows)}") + for caller, akw in rows: + caller_name = caller.get_name() if hasattr(caller, 'get_name') else str(caller) + print(f" Next: {caller_name} with {akw}") + + print("\n" + "=" * 70) + print("Execution Complete") + print("=" * 70) + print(f"\nTotal steps: {step_count - 1}") + print(f"Stashed results: {len(stepper.stash)}") + + if stepper.stash: + print("\nFinal results in stash:") + for i, item in enumerate(stepper.stash, 1): + if hasattr(item, 'get_name'): + node_name = item.get_name() + else: + node_name = str(item) + print(f" {i}. {node_name}") + + +if __name__ == "__main__": + main() diff --git a/workspace/edge-selection.py b/workspace/edge-selection.py new file mode 100644 index 0000000..f3b9a21 --- /dev/null +++ b/workspace/edge-selection.py @@ -0,0 +1,122 @@ +""" +Edge Selection Example - Version 2 + +This is a clean, simple demonstration of custom edge selection using a Unit +subclass with a `get_connections()` method. + +See also: [Version 1](edge-selection-v1.py) - Works but is overly complicated. + +This Version 2 was also written with AI assistance - but after being told off and told +to do it correctly :D + +The key insight: Just override `get_connections()` in your Unit subclass to +filter which connections to follow. The existing infrastructure in edges.py +already checks for and uses this method - no framework changes needed! + +In this example: +- Lana (ChoiceNode) connects to Bob, Alice, Dave1, and Dave2 +- ChoiceNode.get_connections() filters to only "Dave" named connections +- Result: Only Dave1 and Dave2 are called, Bob and Alice are skipped +""" + +from hyperway.graph import Graph +from hyperway.nodes import Unit, as_unit +from hyperway.packer import argspack + + +class ChoiceNode(Unit): + """A custom node that implements get_connections() for edge selection.""" + + def get_connections(self, graph, akw=None): + """Filter and return connections based on custom logic.""" + # Get all connections from graph + connections = graph.get(self.id(), None) + + if not connections: + return None + + name = "bob" + if akw is not None: + print('Using akw for target_name:', akw) + name = akw.args[0].get('target_name', 'alice') + print('Extracted target_name from akw:', name) + else: + print('No akw provided, using default target_name:', name) + + print(f'--- ChoiceNode.get_connections(). Filtering for {name} ---') + # Filter for connections named "Dave" + filtered = tuple(c for c in connections if c.name == name) + names = [c.name for c in filtered] + print(f'--- ChoiceNode.get_connections() filtered connections: {names} ---') + return filtered if filtered else None + + +def print_node(name): + """Factory function to create a print function with a fixed name.""" + def printer(*args, **kwargs): + d = args[0] + if d is None: + print(f" {name} called with no data") + else: + d[name] = True + print(f" {name} called with: args={args}, kwargs={kwargs}") + return d + + printer.__name__ = name + return printer + + +def main(): + """Build and execute the example graph.""" + + # Create the graph + g = Graph() + + # Create the nodes - Lana uses ChoiceNode for custom connection selection + lana = ChoiceNode(print_node("Lana")) + bob_node = as_unit(print_node("Bob")) + alice_node = as_unit(print_node("Alice")) + dave1_node = as_unit(print_node("Dave1")) + dave2_node = as_unit(print_node("Dave2")) + + + # Connect Lana to multiple nodes with named connections + g.add(lana, bob_node, name="bob") + g.add(lana, alice_node, name="alice") + g.add(lana, dave1_node, name="Dave") + g.add(lana, dave2_node, name="Dave") + # g.add(dave1_node, as_unit(print_node("Out")), name="out") + # g.add(dave2_node, as_unit(print_node("Out")), name="out") + + print("=" * 70) + print("Graph Structure:") + print("=" * 70) + print(f"Lana connects to:") + print(f" - Bob (name='bob')") + print(f" - Alice (name='alice')") + print(f" - Dave1 (name='Dave')") + print(f" - Dave2 (name='Dave')") + print() + + # Step the graph from lana until complete + from hyperway import INITIATE_UNIFIED + + print("=" * 70) + print("Executing Graph with initiate=INITIATE_UNIFIED:") + print("=" * 70) + + g.stepper_prepare(lana, {'foo': 'bar', 'target_name': 'Dave'}, initiate=INITIATE_UNIFIED) + stepper = g.stepper() + # return stepper + + while True: + rows = stepper.step() + if not rows: + break + + print("\nExecution complete!") + print(f"Stashed results: {len(stepper.stash)}: {stepper.stash}") + return stepper + +if __name__ == "__main__": + st= main() diff --git a/workspace/initiate-constants-example.py b/workspace/initiate-constants-example.py new file mode 100644 index 0000000..8764b3f --- /dev/null +++ b/workspace/initiate-constants-example.py @@ -0,0 +1,65 @@ +"""Example demonstrating the use of INITIATE_* constants. + +This shows how to use the constants instead of magic strings when +configuring the stepper's initiation mode. +""" + +from hyperway import Graph, as_unit, argspack +from hyperway import INITIATE_DISTRIBUTED, INITIATE_UNIFIED + + +def multiply_by_2(x): + return x * 2 + + +def add_10(x): + return x + 10 + + +def main(): + g = Graph() + + # Create a node with two outgoing connections + start = as_unit(multiply_by_2) + g.add(start, add_10) + g.add(start, add_10) + + print("=" * 60) + print("INITIATE_DISTRIBUTED Mode (default - edge-centric)") + print("=" * 60) + + # Using the constant instead of 'distributed' string + g.stepper_prepare(start, 5) + s = g.stepper() + s.initiate = INITIATE_DISTRIBUTED # Can also be set this way + + rows = s.step() + print(f"Start node called per connection: {len(rows)} rows") + print(f"Initiate mode: {s.initiate}") + + print("\n" + "=" * 60) + print("INITIATE_UNIFIED Mode (node-centric)") + print("=" * 60) + + # Using the constant instead of 'unified' string + s2 = g.stepper() + s2.prepare(start, akw=argspack(5), initiate=INITIATE_UNIFIED) + + rows2 = s2.step() + print(f"Start node called once, result distributed: {len(rows2)} rows") + print(f"Initiate mode: {s2.initiate}") + + print("\n" + "=" * 60) + print("Type-safe constants prevent typos!") + print("=" * 60) + + # These constants are the actual string values + print(f"INITIATE_DISTRIBUTED = '{INITIATE_DISTRIBUTED}'") + print(f"INITIATE_UNIFIED = '{INITIATE_UNIFIED}'") + + # But using constants means typos become NameErrors, not silent bugs + # INITIATE_UNIFFIED would raise NameError instead of silently failing + + +if __name__ == '__main__': + main()