wrap self call to provide helpful info in case of exception plain flow perform shapedtype inference record a new node in the operations graph the node wires itself to inbound and outbound ops the node constructor updates this op s self inboundnodes sets kerashistory on the outputs and adds itself to the outboundnodes of the ops that produced the inputs to this call we override new to saving serializable constructor arguments these arguments are used to autogenerate an object serialization config which enables usercreated subclasses to be serializable out of the box in most cases without forcing the user to manually implement getconfig generate a config to be returned by default by getconfig for safety we only rely on autoconfigs for a small set of serializable types setting an instance attribute in new has the potential to trigger an infinite recursion if a subclass overrides setattr in an unsafe way returns the config of the object an object config is a python dictionary serializable containing the information needed to reinstantiate it in this case the subclass implements getconfig in this case the subclass doesn t implement getconfig let s see if we can autogenerate it remove args non explicitly supported creates a layer from its config this method is the reverse of getconfig capable of instantiating the same layer from the config dictionary it does not handle layer connectivity handled by network nor weights handled by setweights args config a python dictionary typically the output of getconfig returns a layer instance retrieves the input tensors of a symbolic operation only returns the tensors corresponding to the first time the operation was called returns input tensor or list of input tensors retrieves the output tensors of a layer only returns the tensors corresponding to the first time the operation was called returns output tensor or list of output tensors private utility to retrieves an attribute e g inputs from a node this is used to implement the properties output input args nodeindex integer index of the node from which to retrieve the attribute attr exact node attribute name attrname humanreadable attribute name for error messages returns the operation s attribute attr at the node of index nodeindex hooks for backend layer classes can be overridden for per backend post build actions pass def setattrhookself name value wrap self call to provide helpful info in case of exception plain flow perform shape dtype inference record a new node in the operations graph the node wires itself to inbound and outbound ops the node constructor updates this op s self _inbound_nodes sets _keras_history on the outputs and adds itself to the _outbound_nodes of the ops that produced the inputs to this call we override __new__ to saving serializable constructor arguments these arguments are used to auto generate an object serialization config which enables user created subclasses to be serializable out of the box in most cases without forcing the user to manually implement get_config generate a config to be returned by default by get_config for safety we only rely on auto configs for a small set of serializable types setting an instance attribute in __new__ has the potential to trigger an infinite recursion if a subclass overrides setattr in an unsafe way returns the config of the object an object config is a python dictionary serializable containing the information needed to re instantiate it in this case the subclass implements get_config in this case the subclass doesn t implement get_config let s see if we can autogenerate it remove args non explicitly supported object self __class__ __name__ was passing non serializable argument values in __init__ and therefore the object must override get_config in order to be serializable please implement get_config example class customlayer keras layers layer def __init__ self arg1 arg2 kwargs super __init__ kwargs self arg1 arg1 self arg2 arg2 def get_config self config super get_config config update arg1 self arg1 arg2 self arg2 return config creates a layer from its config this method is the reverse of get_config capable of instantiating the same layer from the config dictionary it does not handle layer connectivity handled by network nor weights handled by set_weights args config a python dictionary typically the output of get_config returns a layer instance retrieves the input tensor s of a symbolic operation only returns the tensor s corresponding to the first time the operation was called returns input tensor or list of input tensors retrieves the output tensor s of a layer only returns the tensor s corresponding to the first time the operation was called returns output tensor or list of output tensors private utility to retrieves an attribute e g inputs from a node this is used to implement the properties output input args node_index integer index of the node from which to retrieve the attribute attr exact node attribute name attr_name human readable attribute name for error messages returns the operation s attribute attr at the node of index node_index hooks for backend layer classes can be overridden for per backend post build actions can be overridden for per backend post build actions
import inspect
import textwrap

import tree

from keras import backend
from keras.api_export import keras_export
from keras.backend.common.keras_tensor import any_symbolic_tensors
from keras.ops.node import Node
from keras.utils import python_utils
from keras.utils import traceback_utils
from keras.utils.naming import auto_name


@keras_export("keras.Operation")
class Operation:
    def __init__(self, name=None):
        if name is None:
            name = auto_name(self.__class__.__name__)
        if not isinstance(name, str) or "/" in name:
            raise ValueError(
                "Argument `name` must be a string and "
                "cannot contain character `/`. "
                f"Received: name={name} (of type {type(name)})"
            )
        self.name = name
        self._inbound_nodes = []
        self._outbound_nodes = []

    @traceback_utils.filter_traceback
    def __call__(self, *args, **kwargs):
        if traceback_utils.is_traceback_filtering_enabled():
            
            if any_symbolic_tensors(args, kwargs):
                call_fn = self.symbolic_call
            else:
                call_fn = self.call
            call_fn = traceback_utils.inject_argument_info_in_traceback(
                call_fn,
                object_name=(f"{self.__class__.__name__}.call()"),
            )
            return call_fn(*args, **kwargs)

        
        if any_symbolic_tensors(args, kwargs):
            return self.symbolic_call(*args, **kwargs)
        return self.call(*args, **kwargs)

    def symbolic_call(self, *args, **kwargs):
        
        outputs = self.compute_output_spec(*args, **kwargs)
        
        
        
        
        
        
        Node(
            operation=self, call_args=args, call_kwargs=kwargs, outputs=outputs
        )
        return outputs

    def call(self, *args, **kwargs):
        raise NotImplementedError

    def compute_output_spec(self, *args, **kwargs):
        try:
            return backend.compute_output_spec(self.call, *args, **kwargs)
        except Exception as e:
            if isinstance(e, TypeError):
                raise e
            else:
                new_e = RuntimeError(
                    "Could not automatically infer the output shape / dtype of "
                    f"'{self.name}' (of type {self.__class__.__name__}). "
                    f"Either the `{self.__class__.__name__}.call()` method "
                    f"is incorrect, or you need to implement the "
                    f"`{self.__class__.__name__}.compute_output_spec() / "
                    "compute_output_shape()` method. "
                    f"Error encountered:\n\n{e}"
                )
                raise new_e.with_traceback(e.__traceback__) from None

    def __new__(cls, *args, **kwargs):
        
        
        arg_names = inspect.getfullargspec(cls.__init__).args
        kwargs.update(dict(zip(arg_names[1 : len(args) + 1], args)))
        instance = super(Operation, cls).__new__(cls)
        
        
        supported_types = (str, int, float, bool, type(None))
        try:
            flat_arg_values = tree.flatten(kwargs)
            auto_config = True
            for value in flat_arg_values:
                if not isinstance(value, supported_types):
                    auto_config = False
                    break
        except TypeError:
            auto_config = False
        try:
            instance._lock = False
            if auto_config:
                from keras.saving import serialization_lib

                instance._auto_config = serialization_lib.SerializableDict(
                    **kwargs
                )
            else:
                instance._auto_config = None
            instance._lock = True
        except RecursionError:
            
            
            
            pass
        return instance

    @python_utils.default
    def get_config(self):
        
        config = {
            "name": self.name,
        }

        if not python_utils.is_default(self.get_config):
            
            return config

        
        
        if getattr(self, "_auto_config", None) is not None:
            xtra_args = set(config.keys())
            config.update(self._auto_config.config)
            
            argspec = inspect.getfullargspec(self.__init__)
            if argspec.varkw != "kwargs":
                for key in xtra_args - xtra_args.intersection(argspec.args[1:]):
                    config.pop(key, None)
            return config
        else:
            raise NotImplementedError(
                textwrap.dedent(
                    f
                )
            )

    @classmethod
    def from_config(cls, config):
        
        try:
            return cls(**config)
        except Exception as e:
            raise TypeError(
                f"Error when deserializing class '{cls.__name__}' using "
                f"config={config}.\n\nException encountered: {e}"
            )

    def __repr__(self):
        return f"<Operation name={self.name}>"

    @property
    def input(self):
        
        return self._get_node_attribute_at_index(0, "input_tensors", "input")

    @property
    def output(self):
        
        return self._get_node_attribute_at_index(0, "output_tensors", "output")

    def _get_node_attribute_at_index(self, node_index, attr, attr_name):
        
        if not self._inbound_nodes:
            raise ValueError(
                f"The layer {self.name} has never been called "
                f"and thus has no defined {attr_name}."
            )
        if not len(self._inbound_nodes) > node_index:
            raise ValueError(
                f"Asked to get {attr_name} at node "
                f"{node_index}, but the operation has only "
                f"{len(self._inbound_nodes)} inbound nodes."
            )
        values = getattr(self._inbound_nodes[node_index], attr)
        if isinstance(values, list) and len(values) == 1:
            return values[0]
        else:
            return values

    
    def _post_build(self):
        
        pass

    def _setattr_hook(self, name, value):
        
        return name, value