feat: add ingress decorator for automatic argument transformation with various usage patterns

Author: thorwhalen

i2/castgraph.py

@@ -177,6 +177,9 @@ def transform_func(obj, ctx): ...
 )
 from collections.abc import Callable, Hashable, Iterable, MutableMapping
 
+from i2.wrapper import Wrap
+from i2.signatures import Sig
+
 
 T = TypeVar("T")
 U = TypeVar("U")
@@ -942,6 +945,127 @@ def kinds(self) -> set[Hashable]:
         """
         return self._kinds.copy()
 
+    # -----------------------------
+    # Ingress decorator for automatic argument transformation
+    # -----------------------------
+
+    @property
+    def ingress(self):
+        """Return decorator factory with attribute access for kinds.
+
+        This property provides a flexible interface for decorating functions to
+        automatically transform their arguments to specified kinds.
+
+        Usage patterns:
+
+        1. Specify kind and argument name:
+           @graph.ingress('text', 'content')
+           def func(content): ...
+
+        2. Specify kind only (transforms first argument):
+           @graph.ingress('text')
+           def func(arg): ...
+
+        3. Use keyword argument:
+           @graph.ingress(arg_name='text')
+           def func(arg_name): ...
+
+        4. Attribute-based syntax for registered kinds:
+           @graph.ingress.text('content')
+           def func(content): ...
+
+        5. Attribute-based for first argument:
+           @graph.ingress.text
+           def func(arg): ...
+
+        Examples
+        --------
+        >>> graph = TransformationGraph()
+        >>> graph.add_node('text', isa=lambda x: isinstance(x, str))
+        >>> graph.add_node(int)
+        >>> @graph.register_edge('text', int)
+        ... def text_to_int(s, ctx): return int(s)
+        >>> @graph.ingress('text')
+        ... def process(x):
+        ...     return x + ' processed'
+        >>> # Can now pass int, will be transformed to text first
+        """
+        return _IngressProxy(self)
+
+    def _ingress_decorator(
+        self,
+        kind_or_arg: Hashable | None = None,
+        arg_name: str | None = None,
+        *,
+        context: dict | None = None,
+    ):
+        """Internal method to create ingress decorator.
+
+        Parameters
+        ----------
+        kind_or_arg : Hashable | None
+            The target kind for transformation, or argument name if arg_name is provided
+        arg_name : str | None
+            The name of the argument to transform. If None, transforms first argument.
+        context : dict | None
+            Optional context to pass to transformations
+
+        Returns
+        -------
+        Callable
+            Decorator function
+        """
+        # Determine target kind and argument name
+        if arg_name is not None:
+            # @graph.ingress(str, 'x') or @graph.ingress('text', 'x')
+            target_kind = kind_or_arg
+            target_arg = arg_name
+        elif isinstance(kind_or_arg, str) or isinstance(kind_or_arg, type):
+            # @graph.ingress('text') or @graph.ingress(int)
+            target_kind = kind_or_arg
+            target_arg = None  # Will use first arg
+        else:
+            # kind_or_arg could be None or some other hashable
+            target_kind = kind_or_arg
+            target_arg = None
+
+        def decorator(func):
+            nonlocal target_arg
+            sig = Sig(func)
+
+            # Default to first arg if not specified
+            if target_arg is None:
+                if not sig.names:
+                    raise ValueError(
+                        f"Function {func.__name__} has no parameters to transform"
+                    )
+                target_arg = sig.names[0]
+
+            # Validate target_arg exists
+            if target_arg not in sig.names:
+                raise ValueError(
+                    f"Argument '{target_arg}' not found in function {func.__name__}. "
+                    f"Available arguments: {sig.names}"
+                )
+
+            # Create ingress function
+            def ingress_func(*args, **kwargs):
+                # Map to all kwargs
+                all_kwargs = sig.map_arguments(args, kwargs, apply_defaults=False)
+
+                # Transform the target argument if present
+                if target_arg in all_kwargs:
+                    all_kwargs[target_arg] = self.transform_any(
+                        all_kwargs[target_arg], target_kind, context=context
+                    )
+
+                # Convert back to args/kwargs respecting signature
+                return sig.mk_args_and_kwargs(all_kwargs, allow_partial=True)
+
+            return Wrap(func, ingress=ingress_func)
+
+        return decorator
+
     # -----------------------------
     # Backward compatibility methods (deprecated)
     # -----------------------------
@@ -1558,6 +1682,69 @@ def _find_min_cost_path(self, src: type[Any], dst: type[Any]) -> list[Edge]:
 # ----------------------------------------------------------------------
 
 
+class _IngressProxy:
+    """Helper class to provide attribute-based access to kinds for ingress decorator.
+
+    This class enables syntax like @graph.ingress.text or @graph.ingress.int
+    by dynamically looking up kinds and creating decorators.
+    """
+
+    def __init__(self, graph: TransformationGraph):
+        self._graph = graph
+
+    def __call__(
+        self,
+        kind_or_arg: Hashable | None = None,
+        arg_name: str | None = None,
+        *,
+        context: dict | None = None,
+    ):
+        """Allow calling as @graph.ingress(kind, arg_name)."""
+        return self._graph._ingress_decorator(kind_or_arg, arg_name, context=context)
+
+    def __getattr__(self, kind_name: str):
+        """Enable attribute access like @graph.ingress.text or @graph.ingress.int.
+
+        Looks up the kind by string name or by type.__name__.
+        Returns a decorator or a decorator factory depending on usage.
+        """
+        # Look up kind by string name or type.__name__
+        matching_kind = None
+        for kind in self._graph.kinds():
+            if isinstance(kind, str) and kind == kind_name:
+                matching_kind = kind
+                break
+            elif isinstance(kind, type) and kind.__name__ == kind_name:
+                matching_kind = kind
+                break
+
+        if matching_kind is None:
+            raise AttributeError(
+                f"Kind '{kind_name}' not found in graph. "
+                f"Available kinds: {self._graph.kinds()}"
+            )
+
+        # Return a factory that can be used as @graph.ingress.kind or @graph.ingress.kind(arg_name)
+        return _KindIngressFactory(self._graph, matching_kind)
+
+
+class _KindIngressFactory:
+    """Factory to handle @graph.ingress.kind and @graph.ingress.kind(arg_name) syntax."""
+
+    def __init__(self, graph: TransformationGraph, kind: Hashable):
+        self._graph = graph
+        self._kind = kind
+
+    def __call__(self, func_or_arg_name):
+        """Handle both @graph.ingress.kind and @graph.ingress.kind(arg_name) patterns."""
+        if callable(func_or_arg_name):
+            # Used as @graph.ingress.kind (no parentheses, applied to first arg)
+            return self._graph._ingress_decorator(self._kind, None)(func_or_arg_name)
+        else:
+            # Used as @graph.ingress.kind(arg_name) or @graph.ingress.kind('arg')
+            return self._graph._ingress_decorator(self._kind, func_or_arg_name)
+
+
 def design_guidelines() -> str:
     """
     Returns concise guidance for organizing casting in Python.

i2/tests/test_castgraph.py

@@ -332,3 +332,83 @@ class M2: ...
         @reg.register()
         def bad(x):
             return None
+
+
+# --- Tests for ingress decorator ---
+
+
+def test_ingress_with_string_kinds():
+    """Test ingress decorator with string-based kinds."""
+    from i2.castgraph import TransformationGraph
+
+    graph = TransformationGraph()
+    graph.add_node('text', isa=lambda x: isinstance(x, str))
+    graph.add_node('number', isa=lambda x: isinstance(x, (int, float)))
+
+    @graph.register_edge('text', 'number')
+    def text_to_number(t, ctx):
+        return float(t)
+
+    @graph.ingress('number', 'x')
+    def double(x):
+        return x * 2
+
+    result = double("21")
+    assert result == 42.0
+
+
+def test_ingress_with_type_kinds():
+    """Test ingress decorator with type-based kinds."""
+    from i2.castgraph import TransformationGraph
+
+    graph = TransformationGraph()
+
+    @graph.register_edge(str, int)
+    def str_to_int(s, ctx):
+        return int(s)
+
+    @graph.ingress(int)
+    def square(n):
+        return n**2
+
+    result = square("5")
+    assert result == 25
+
+
+def test_ingress_attribute_syntax():
+    """Test ingress decorator with attribute-based syntax."""
+    from i2.castgraph import TransformationGraph
+
+    graph = TransformationGraph()
+    graph.add_node('data', isa=lambda x: isinstance(x, dict))
+
+    @graph.register_edge(str, 'data')
+    def str_to_data(s, ctx):
+        return json.loads(s)
+
+    @graph.ingress.data('obj')
+    def get_value(obj):
+        return obj.get('value', 0)
+
+    result = get_value('{"value": 42}')
+    assert result == 42
+
+
+def test_ingress_preserves_function_signature():
+    """Test that ingress decorator preserves function behavior."""
+    from i2.castgraph import TransformationGraph
+
+    graph = TransformationGraph()
+
+    @graph.register_edge(str, int)
+    def str_to_int(s, ctx):
+        return int(s)
+
+    @graph.ingress(int, 'a')
+    def add(a, b):
+        return a + b
+
+    # Test with various call patterns
+    assert add("10", 5) == 15
+    assert add(a="20", b=22) == 42
+    assert add("7", b=8) == 15