Enhance functionality with new features and error handling

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "tests"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

docs/kwargs-bubbling.md

@@ -0,0 +1,121 @@
+# Kwargs Bubbling
+
+Wire functions can pass metadata through the graph without polluting function signatures. This is _kwargs bubbling_ - setup kwargs flow through the pipeline in the `ArgsPack`, independent of runtime function calls.
+
+## The Problem
+
+Consider passing metadata through a graph pipeline:
+
+```py
+def add(x, y, **metadata):
+    # Function forced to accept **metadata
+    # even though it doesn't use it
+    return x + y
+
+def multiply(value, **metadata):
+    # Every function needs **metadata
+    return value * 2
+```
+
+This is noisy. Functions have to accept kwargs they don't use.
+
+## Kwargs Bubbling
+
+Wire functions separate runtime kwargs from pipeline metadata:
+
+```py
+from hyperway.edges import wire
+
+def add(x, y):
+    # Clean signature
+    return x + y
+
+def multiply(value):
+    # No **kwargs noise
+    return value * 2
+
+# Setup wire with pipeline metadata
+wire_add = wire(add, stage='sum', pipeline='data-transform')
+wire_mul = wire(multiply, stage='amplify')
+```
+
+When called:
+
+```py
+result = wire_add(10, 20)
+# Function receives: add(10, 20)
+# ArgsPack contains: (30, stage='sum', pipeline='data-transform')
+```
+
+The metadata _bubbles_ through the `ArgsPack`, never touching the function signature.
+
+## Call-time vs Setup-time
+
+Two distinct flows:
+
+**Call-time kwargs** → Go to the function
+```py
+def add_with_multiplier(x, y, multiplier=1):
+    return (x + y) * multiplier
+
+wire_func = wire(add_with_multiplier, pipeline='test')
+result = wire_func(10, 20, multiplier=5)
+# Function receives: add_with_multiplier(10, 20, multiplier=5)
+# ArgsPack contains: (150, pipeline='test')
+```
+
+**Setup-time kwargs** → Bubble through `ArgsPack`
+```py
+wire_func = wire(add, stage='sum', meta='important')
+result = wire_func(10, 20)
+# Function receives: add(10, 20)
+# ArgsPack contains: (30, stage='sum', meta='important')
+```
+
+## Usage in Graphs
+
+This keeps graph pipelines clean while preserving metadata for debugging, logging, or conditional execution:
+
+```py
+from hyperway.edges import make_edge
+
+g = Graph()
+
+# Each wire carries stage metadata
+edge1 = g.add(node_a, node_b, through=wire(transform, stage='normalize'))
+edge2 = g.add(node_b, node_c, through=wire(validate, stage='check'))
+
+# Metadata flows through without touching function signatures
+g.stepper_prepare(node_a, initial_data)
+stepper = g.stepper()
+
+for rows in stepper:
+    for caller, argpack in rows:
+        # Access bubbled metadata
+        print(f"Stage: {argpack.kwargs.get('stage')}")
+```
+
+## wire vs wire_partial
+
+Both support kwargs bubbling:
+
+**`wire`** - No pre-applied args, just metadata:
+```py
+wire_func = wire(my_function, pipeline='prod', version='2.0')
+result = wire_func(x, y)  # Clean call
+```
+
+**`wire_partial`** - Pre-applied args AND result goes to function:
+```py
+wire_func = wire_partial(my_function, preset_arg, foo=2)
+result = wire_func(runtime_arg, foo=5)  # foo=5 overrides foo=2
+# Function receives all merged kwargs
+```
+
+## Key Insight
+
+Kwargs bubbling separates concerns:
+- **Function signature** = What the function needs to compute
+- **Pipeline metadata** = What the graph needs to track
+
+Functions stay clean. Metadata flows through. The graph stays debuggable.

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hyperway"
-version = "1.0.4"
+version = "1.0.5"
 authors = [
   { name="Strangemother", email="hyperway@strangemother.com" },
 ]

sonar-project.properties

@@ -5,7 +5,7 @@
 sonar.projectKey=Strangemother_python-hyperway
 sonar.organization=strangemother
 sonar.projectName=Python Hyperway
-sonar.projectVersion=1.0.4
+sonar.projectVersion=1.0.5
 
 # Source code settings
 sonar.sources=src/hyperway

src/hyperway/__init__.py

@@ -6,4 +6,5 @@
 from .nodes import as_unit, as_units, Unit
 from .edges import make_edge, Connection
 from .graph import Graph
-from .packer import argspack, argpack, ArgsPack
+from .packer import argspack, argpack, ArgsPack
+from .stepper import StepperException

src/hyperway/edges.py

@@ -15,7 +15,36 @@ def is_edge(unit):
     return isinstance(unit, Connection)
 
 
-def wire(func, *a, **kw):
+def wire_partial(func, *wire_a, **wire_kw):
+    """Wire function that accepts a standard function, to 
+    later execute as a wire function.
+
+    This _partial_ variant acts like a funtools.partial, allowing
+    pre-setting of arguments to the wire function.
+
+        def my_wire_func(*a, **kw):
+            print('WIRING:', a, kw)
+            return argspack(*a, **kw)
+
+        wire_partial(my_wire_func, 10, 20, foo=2, egg=True)
+        my_wire_func(30, 40, foo=5)
+        WIRING: (10, 20, 30, 40) {'foo': 5, 'egg': True}
+
+    Allowing the developer to apply generic functions as a wire
+    method without handling the argpack.
+
+        c = make_edge(f.add_1, f.add_2, 
+                through=wire(f.mul_5))
+    """
+    def wrapper(*a, **kw):
+        """Wrapper wire function."""
+        wire_kw.update(kw)
+        result = func(*wire_a, *a, **wire_kw)
+        return argspack(result)
+    return wrapper
+
+
+def wire(func, **wkw):
     """Wire function that accepts a standard function, to 
     later execute as a wire function
 
@@ -24,11 +53,15 @@ def wire(func, *a, **kw):
 
         c = make_edge(f.add_1, f.add_2, 
                 through=wire(f.mul_5))
+
+    The wrapped function is called with all runtime args/kwargs, allowing
+    it to work with its normal signature. Wire-setup kwargs (wkw) flow 
+    through the ArgsPack for pipeline metadata.
     """
     def wrapper(*a, **kw):
         """Wrapper wire function."""
         result = func(*a, **kw)
-        return argspack(result, **kw)
+        return argspack(result, **wkw)
     return wrapper
 
 
@@ -105,8 +138,8 @@ def __repr__(self):
     def __call__(self, *a, _graph=None, **kw):
         """Call this connection, returning the result of B.
         This will call A, then the through function (if any), then B."""
-        g = _graph or self.on
-        return self.get_a().process(*a, **kw)
+        g = self.on if _graph is None else _graph
+        return self.get_a(g).process(*a, **kw)
 
     @property
     def merge_node(self):

src/hyperway/graph/base.py

@@ -18,7 +18,9 @@ def resolve_node_connections(self, other):
             return self.resolve_node_to_nowhere(other)
         return res
 
-    def resolve_node_to_nowhere(self, other):
+    def resolve_node_to_nowhere(self, other): # py-lint: disable=unused-argument
+        """Given a node with no connections, return a connection to nowhere.
+        Override in subclasses to provide custom behaviour."""
         return ()
 
 

src/hyperway/packer.py

@@ -67,7 +67,7 @@ def argpack(result=UNDEFINED, *more, **extra):
 def test_argpack():
 
     a = (1,3,4,5)
-    d = dict(foo=3, bar=4)
+    d = {'foo': 3, 'bar': 4}
 
     v = argpack(argpack(a, **d))
     assert a == v.args[0]

src/hyperway/reader.py

@@ -11,7 +11,7 @@ def thin_graph(graph):
 
 
 def flat_graph(graph):
-    res = tuple()
+    res = ()
     thin = thin_graph(graph)
     for node_a, b_nodes in thin.items():
         for node_b in b_nodes:

src/hyperway/stepper.py

@@ -7,6 +7,12 @@
                     is_edge, PartialConnection, get_connections)
 from .graph.base import is_graph
 
+
+class StepperException(Exception):
+    """Exception raised when stepper encounters an error during execution."""
+    pass
+
+
 def run_stepper(g, unit_a, val):
     """Here we can _run_ the graph, executing the chain of connections
     from A to Z. The _result_ is the end attribute. This may be a graph of
@@ -93,6 +99,7 @@ def set_global_expand(expand_func):
     global expand
     expand = expand_func    
 
+
 def stepper_c(graph, start_node, argspack):
     stepper = StepperC(graph)
     res = stepper.start(start_node, akw=argspack)
@@ -191,7 +198,7 @@ def step(self, rows=None, count=1):
         st_nodes = self.start_nodes
         if st_nodes is None:
             # Start node must be something...
-            raise Exception('start_nodes is None')
+            raise StepperException('start_nodes is None')
         self.rows = rows or self.rows or expand(st_nodes, self.start_akw,)
 
         while c < count:
@@ -364,7 +371,7 @@ def call_one(self, func, akw):
 
         return self.call_one_fallthrough(func, akw)
 
-    def call_one_fallthrough(self, thing, akw):
+    def call_one_fallthrough(self, thing, akw): # pylint: disable=unused-argument
         """
         The given function is not A Connection, PartialConnection, Unit, or
         function (a callable). The last-stage action should occur.
@@ -467,6 +474,12 @@ def no_branch(self, func, akw):
         return self.end_branch(func, akw)
 
     def end_branch(self, func, akw):
+        """The end_branch method is default hanlder when no branches exist on the stepper chain.
+        This captures the result from the last call, and stores it in the stash for later retrieval.
+
+        return an empty tuple if the stash branch stash ends, else return a tuple of tuples (a row set)
+        with no destination node. 
+        """
         # print(' ... Connections end ...', akw)
         # A tuple of rows
         if self.stash_ends:

src/hyperway/writer.py

@@ -42,6 +42,8 @@ def write_graphviz(graph, title, **opts):
 
     styles = opts.pop('styles', {})
     direction = opts.pop('direction', 'TB')
+    show_view = opts.pop('view', False)
+    t_format = opts.pop('format', None)
     print(f'{direction=}')
     defaults = {
         # 'format':'svg',
@@ -85,8 +87,10 @@ def write_graphviz(graph, title, **opts):
     t.attr(fontsize=styles.get('fontsize', '12'))
     t.attr(bgcolor=styles.get('bgcolor', "#00000000"))
 
-    # t.view()
-    # t.format = 'svg'
+    if show_view:
+        t.view()
+    if t_format is not None:
+        t.format = t_format
     r_opts = {}
     directory = opts.get('directory', None)
     if directory is not None: