broadcast input shapes to a unified shape convert to list for mutability args shape1 a tuple or list of integers shape2 a tuple or list of integers returns outputshape list of integers or none the broadcasted shape example broadcastshapes5 3 1 3 5 3 compute the output shape for the expanddims operation args inputshape input shape axis int for the axis to expand returns tuple of ints the output shape after the expanddims operation computes the output shape of pooling operations args inputshape input shape must be a tuple of integers poolsize size of the pooling operation must be a tuple of integers strides stride of the pooling operation must be a tuple of integers defaults to poolsize padding padding method available methods are valid or same defaults to valid dataformat string either channelslast or channelsfirst the ordering of the dimensions in the inputs channelslast corresponds to inputs with shape batch height width channels while channelsfirst corresponds to inputs with shape batch channels height weight defaults to channelslast returns tuple of ints the output shape of the pooling operation examples basic usage with square pooling on a single image computepoolingoutputshape1 4 4 1 2 2 1 2 2 1 strided pooling on a single image with strides different from poolsize computepoolingoutputshape1 4 4 1 2 2 strides1 1 1 3 3 1 pooling on a batch of images computepoolingoutputshape32 4 4 3 2 2 32 2 2 3 set none shape to a manual value so that we can run numpy computation on spatialshape compute the output shape of conv ops if dataformat channelslast spatialshape inputshape1 1 kernelshape kernelsize inputshape1 filters else spatialshape inputshape2 kernelshape kernelsize inputshape1 filters if lenkernelshape leninputshape raise valueerror kernel shape must have the same length as input but received fkernel of shape kernelshape and finput of shape inputshape if isinstancedilationrate int dilationrate dilationrate lenspatialshape if isinstancestrides int strides strides lenspatialshape if lendilationrate lenspatialshape raise valueerror dilation must be none scalar or tuplelist of length of inputs spatial shape but received fdilationratedilationrate and finput of shape inputshape nonedims spatialshape np arrayspatialshape for i in rangelenspatialshape if spatialshapei is none set none shape to a manual value so that we can run numpy computation on spatialshape spatialshapei 1 nonedims appendi kernelspatialshape np arraykernelshape 2 dilationrate np arraydilationrate if padding valid outputspatialshape np floor spatialshape dilationrate kernelspatialshape 1 1 strides 1 for i in rangelenoutputspatialshape if i not in nonedims and outputspatialshapei 0 raise valueerror computed output size would be negative received finputs shapeinputshape fkernel shapekernelshape fdilationratedilationrate elif padding same or padding causal outputspatialshape np floorspatialshape 1 strides 1 outputspatialshape inti for i in outputspatialshape for i in nonedims outputspatialshapei none outputspatialshape tupleoutputspatialshape if dataformat channelslast outputshape inputshape0 outputspatialshape kernelshape1 else outputshape inputshape0 kernelshape1 outputspatialshape return outputshape def computematmuloutputshapeshape1 shape2 if lenshape1 1 shape1 1 shape10 if lenshape2 1 shape2 shape20 1 if shape11 is not none and shape22 is not none and shape11 shape22 raise valueerror inner dimensions x1 shape1 and x2 shape2 must be fequal but received x1 shapeshape1 and fx2 shapeshape2 leadingshape broadcastshapesshape1 2 shape2 2 last2dimsshape shape12 shape21 outputshape leadingshape last2dimsshape if lenshape1 1 del outputshape2 if lenshape2 1 del outputshape1 return tupleoutputshape def computereshapeoutputshapeinputshape newshape newshapeargname unknowndimcount newshape count1 if unknowndimcount 1 raise valueerror there must be at most one unknown dimension 1 in fnewshapeargname received newshapeargnamenewshape if there is a none in inputshape we can t infer what the 1 is if none in inputshape return tupledim if dim 1 else none for dim in newshape inputsize math prodinputshape if the newshape fully defined return it if unknowndimcount 0 if inputsize math prodnewshape raise valueerror the total size of the tensor must be unchanged received finputshapeinputshape newshapeargnamenewshape return newshape we have one 1 in newshape compute the actual value knownoutputsize 1 unknowndimindex none for index dim in enumeratenewshape if dim 1 unknowndimindex index else knownoutputsize dim if knownoutputsize 0 or inputsize knownoutputsize 0 raise valueerror the total size of the tensor must be unchanged however the input size cannot by divided by the specified dimensions in fnewshapeargname received inputshapeinputshape fnewshapeargnamenewshape outputshape listnewshape outputshapeunknowndimindex inputsize knownoutputsize return tupleoutputshape def computetransposeoutputshapeinputshape axes inputshape listinputshape if axes is none return tupleinputshape 1 if lenaxes leninputshape raise valueerror axis must be a list of the same length as the input shape fexpected leninputshape but received lenaxes return tupleinputshapeax for ax in axes def reduceshapeshape axisnone keepdimsfalse shape listshape if axis is none if keepdims return tuple1 for in shape else return tuple if keepdims for ax in axis shapeax 1 return tupleshape else for ax in sortedaxis reversetrue del shapeax return tupleshape kerasexportkeras utils getsourceinputs def getsourceinputstensor if not hasattrtensor kerashistory return tensor operation nodeindex tensor kerashistory if not operation or not operation inboundnodes return tensor else node operation inboundnodesnodeindex if node isinput reached input node stop recursion return tree flattennode outputtensors else sourcetensors for tensor in node inputtensors previoussources getsourceinputstensor avoid input redundancy for x in previoussources if allx is not t for t in sourcetensors sourcetensors appendx return sourcetensors broadcast input shapes to a unified shape convert to list for mutability args shape1 a tuple or list of integers shape2 a tuple or list of integers returns output_shape list of integers or none the broadcasted shape example broadcast_shapes 5 3 1 3 5 3 compute the output shape for the expand_dims operation args input_shape input shape axis int for the axis to expand returns tuple of ints the output shape after the expand_dims operation computes the output shape of pooling operations args input_shape input shape must be a tuple of integers pool_size size of the pooling operation must be a tuple of integers strides stride of the pooling operation must be a tuple of integers defaults to pool_size padding padding method available methods are valid or same defaults to valid data_format string either channels_last or channels_first the ordering of the dimensions in the inputs channels_last corresponds to inputs with shape batch height width channels while channels_first corresponds to inputs with shape batch channels height weight defaults to channels_last returns tuple of ints the output shape of the pooling operation examples basic usage with square pooling on a single image compute_pooling_output_shape 1 4 4 1 2 2 1 2 2 1 strided pooling on a single image with strides different from pool_size compute_pooling_output_shape 1 4 4 1 2 2 strides 1 1 1 3 3 1 pooling on a batch of images compute_pooling_output_shape 32 4 4 3 2 2 32 2 2 3 set none shape to a manual value so that we can run numpy computation on spatial_shape compute the output shape of conv ops set none shape to a manual value so that we can run numpy computation on spatial_shape compute the output shape of a matmul operation args shape1 shape of the left operand shape2 shape of the right operand returns tuple of ints the output shape for the matmul operation converts 1 in new_shape to either an actual dimension or none this utility does not special case the 0th dimension batch size if there is a none in input_shape we can t infer what the 1 is if the new_shape fully defined return it we have one 1 in new_shape compute the actual value compute the output shape for the transpose operation args input_shape input shape axes permutation of the dimensions for the transpose operation returns tuple of ints the output shape after the transpose operation returns the list of input tensors necessary to compute tensor output will always be a list of tensors potentially with 1 element args tensor the tensor to start from returns list of input tensors reached input node stop recursion avoid input redundancy
import math

import numpy as np
import tree

from keras.api_export import keras_export


def broadcast_shapes(shape1, shape2):
    
    shape1 = list(shape1)
    shape2 = list(shape2)
    origin_shape1 = shape1
    origin_shape2 = shape2

    if len(shape1) > len(shape2):
        shape2 = [1] * (len(shape1) - len(shape2)) + shape2
    if len(shape1) < len(shape2):
        shape1 = [1] * (len(shape2) - len(shape1)) + shape1
    output_shape = list(shape1)
    for i in range(len(shape1)):
        if shape1[i] == 1:
            output_shape[i] = shape2[i]
        elif shape1[i] is None:
            output_shape[i] = None if shape2[i] == 1 else shape2[i]
        else:
            if shape2[i] == 1 or shape2[i] is None or shape2[i] == shape1[i]:
                output_shape[i] = shape1[i]
            else:
                raise ValueError(
                    "Cannot broadcast shape, the failure dim has value "
                    f"{shape1[i]}, which cannot be broadcasted to {shape2[i]}. "
                    f"Input shapes are: {origin_shape1} and {origin_shape2}."
                )

    return output_shape


def compute_expand_dims_output_shape(input_shape, axis):
    
    input_shape = list(input_shape)
    if axis is None:
        axis = len(input_shape)
    elif axis < 0:
        axis = len(input_shape) + 1 + axis
    return tuple(input_shape[:axis] + [1] + input_shape[axis:])


def compute_pooling_output_shape(
    input_shape,
    pool_size,
    strides,
    padding="valid",
    data_format="channels_last",
):
    
    strides = pool_size if strides is None else strides
    input_shape_origin = list(input_shape)
    input_shape = np.array(input_shape)
    if data_format == "channels_last":
        spatial_shape = input_shape[1:-1]
    else:
        spatial_shape = input_shape[2:]
    none_dims = []
    for i in range(len(spatial_shape)):
        if spatial_shape[i] is None:
            
            
            spatial_shape[i] = -1
            none_dims.append(i)
    pool_size = np.array(pool_size)
    if padding == "valid":
        output_spatial_shape = (
            np.floor((spatial_shape - pool_size) / strides) + 1
        )
        for i in range(len(output_spatial_shape)):
            if i not in none_dims and output_spatial_shape[i] < 0:
                raise ValueError(
                    "Computed output size would be negative. Received: "
                    f"`inputs.shape={input_shape}` and `pool_size={pool_size}`."
                )
    elif padding == "same":
        output_spatial_shape = np.floor((spatial_shape - 1) / strides) + 1
    else:
        raise ValueError(
            "`padding` must be either `'valid'` or `'same'`. Received "
            f"{padding}."
        )
    output_spatial_shape = [int(i) for i in output_spatial_shape]
    for i in none_dims:
        output_spatial_shape[i] = None
    output_spatial_shape = tuple(output_spatial_shape)
    if data_format == "channels_last":
        output_shape = (
            (input_shape_origin[0],)
            + output_spatial_shape
            + (input_shape_origin[-1],)
        )
    else:
        output_shape = (
            input_shape_origin[0],
            input_shape_origin[1],
        ) + output_spatial_shape
    return output_shape


def compute_conv_output_shape(
    input_shape,
    filters,
    kernel_size,
    strides=1,
    padding="valid",
    data_format="channels_last",
    dilation_rate=1,
):
    
    if data_format == "channels_last":
        spatial_shape = input_shape[1:-1]
        kernel_shape = kernel_size + (input_shape[-1], filters)
    else:
        spatial_shape = input_shape[2:]
        kernel_shape = kernel_size + (input_shape[1], filters)
    if len(kernel_shape) != len(input_shape):
        raise ValueError(
            "Kernel shape must have the same length as input, but received "
            f"kernel of shape {kernel_shape} and "
            f"input of shape {input_shape}."
        )
    if isinstance(dilation_rate, int):
        dilation_rate = (dilation_rate,) * len(spatial_shape)
    if isinstance(strides, int):
        strides = (strides,) * len(spatial_shape)
    if len(dilation_rate) != len(spatial_shape):
        raise ValueError(
            "Dilation must be None, scalar or tuple/list of length of "
            "inputs' spatial shape, but received "
            f"`dilation_rate={dilation_rate}` and "
            f"input of shape {input_shape}."
        )
    none_dims = []
    spatial_shape = np.array(spatial_shape)
    for i in range(len(spatial_shape)):
        if spatial_shape[i] is None:
            
            
            spatial_shape[i] = -1
            none_dims.append(i)

    kernel_spatial_shape = np.array(kernel_shape[:-2])
    dilation_rate = np.array(dilation_rate)
    if padding == "valid":
        output_spatial_shape = (
            np.floor(
                (spatial_shape - dilation_rate * (kernel_spatial_shape - 1) - 1)
                / strides
            )
            + 1
        )
        for i in range(len(output_spatial_shape)):
            if i not in none_dims and output_spatial_shape[i] < 0:
                raise ValueError(
                    "Computed output size would be negative. Received "
                    f"`inputs shape={input_shape}`, "
                    f"`kernel shape={kernel_shape}`, "
                    f"`dilation_rate={dilation_rate}`."
                )
    elif padding == "same" or padding == "causal":
        output_spatial_shape = np.floor((spatial_shape - 1) / strides) + 1
    output_spatial_shape = [int(i) for i in output_spatial_shape]
    for i in none_dims:
        output_spatial_shape[i] = None
    output_spatial_shape = tuple(output_spatial_shape)
    if data_format == "channels_last":
        output_shape = (
            (input_shape[0],) + output_spatial_shape + (kernel_shape[-1],)
        )
    else:
        output_shape = (input_shape[0], kernel_shape[-1]) + output_spatial_shape
    return output_shape


def compute_matmul_output_shape(shape1, shape2):
    
    if len(shape1) == 1:
        shape1 = (1, shape1[0])
    if len(shape2) == 1:
        shape2 = (shape2[0], 1)
    if (
        shape1[-1] is not None
        and shape2[-2] is not None
        and shape1[-1] != shape2[-2]
    ):
        raise ValueError(
            "Inner dimensions (`x1.shape[-1]` and `x2.shape[-2]`) must be "
            f"equal, but received `x1.shape={shape1}` and "
            f"`x2.shape={shape2}`."
        )

    leading_shape = broadcast_shapes(shape1[:-2], shape2[:-2])
    last_2_dims_shape = [shape1[-2], shape2[-1]]
    output_shape = leading_shape + last_2_dims_shape
    if len(shape1) == 1:
        del output_shape[-2]
    if len(shape2) == 1:
        del output_shape[-1]
    return tuple(output_shape)


def compute_reshape_output_shape(input_shape, new_shape, new_shape_arg_name):
    
    unknown_dim_count = new_shape.count(-1)
    if unknown_dim_count > 1:
        raise ValueError(
            "There must be at most one unknown dimension (-1) in "
            f"{new_shape_arg_name}. Received: {new_shape_arg_name}={new_shape}."
        )

    
    if None in input_shape:
        return tuple(dim if dim != -1 else None for dim in new_shape)

    input_size = math.prod(input_shape)
    
    if unknown_dim_count == 0:
        if input_size != math.prod(new_shape):
            raise ValueError(
                "The total size of the tensor must be unchanged. Received: "
                f"input_shape={input_shape}, {new_shape_arg_name}={new_shape}"
            )
        return new_shape

    
    known_output_size = 1
    unknown_dim_index = None
    for index, dim in enumerate(new_shape):
        if dim == -1:
            unknown_dim_index = index
        else:
            known_output_size *= dim

    if known_output_size == 0 or input_size % known_output_size != 0:
        raise ValueError(
            "The total size of the tensor must be unchanged, however, the "
            "input size cannot by divided by the specified dimensions in "
            f"{new_shape_arg_name}. Received: input_shape={input_shape}, "
            f"{new_shape_arg_name}={new_shape}"
        )

    output_shape = list(new_shape)
    output_shape[unknown_dim_index] = input_size // known_output_size
    return tuple(output_shape)


def compute_transpose_output_shape(input_shape, axes):
    
    input_shape = list(input_shape)
    if axes is None:
        return tuple(input_shape[::-1])

    if len(axes) != len(input_shape):
        raise ValueError(
            "axis must be a list of the same length as the input shape, "
            f"expected {len(input_shape)}, but received {len(axes)}."
        )
    return tuple(input_shape[ax] for ax in axes)


def reduce_shape(shape, axis=None, keepdims=False):
    shape = list(shape)
    if axis is None:
        if keepdims:
            return tuple([1 for _ in shape])
        else:
            return tuple([])

    if keepdims:
        for ax in axis:
            shape[ax] = 1
        return tuple(shape)
    else:
        for ax in sorted(axis, reverse=True):
            del shape[ax]
        return tuple(shape)


@keras_export("keras.utils.get_source_inputs")
def get_source_inputs(tensor):
    
    if not hasattr(tensor, "_keras_history"):
        return tensor

    operation, node_index, _ = tensor._keras_history
    if not operation or not operation._inbound_nodes:
        return [tensor]
    else:
        node = operation._inbound_nodes[node_index]
        if node.is_input:
            
            return tree.flatten(node.output_tensors)
        else:
            source_tensors = []
            for tensor in node.input_tensors:
                previous_sources = get_source_inputs(tensor)
                
                for x in previous_sources:
                    if all(x is not t for t in source_tensors):
                        source_tensors.append(x)
            return source_tensors