diff --git a/test/support/circle/README.md b/test/support/circle/README.md new file mode 100644 index 00000000..3fa9afad --- /dev/null +++ b/test/support/circle/README.md @@ -0,0 +1,23 @@ +# Circle value-test support + +This directory contains test-only infrastructure for checking numerical equivalence of +Circle-to-Circle rewrites without `circle-interpreter` or `onert`. + +## Components + +- `builder.py` creates small, serializable `circle.ModelT` fixtures. +- `evaluator.py` evaluates a deliberately limited operator subset with NumPy and records + every intermediate tensor value. +- `value_test.py` provides reusable assertions for serialization round trips, pass + equivalence, graph-interface preservation, and extraction-boundary equivalence. + +The reference evaluator currently supports `ADD`, `SUB`, `MUL`, `RESHAPE`, and +`TRANSPOSE`. Unsupported operators, fused activations, optional inputs, external +buffers, and unsupported tensor types fail explicitly. The evaluator is not a +production Circle runtime. + +Run the value tests with: + +```bash +./ccex test -k circle.value +``` diff --git a/test/support/circle/__init__.py b/test/support/circle/__init__.py new file mode 100644 index 00000000..6598579f --- /dev/null +++ b/test/support/circle/__init__.py @@ -0,0 +1,45 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Support utilities for execution-based Circle value tests.""" + +from test.support.circle.builder import CircleModelBuilder +from test.support.circle.evaluator import ( + circle_tensor_type_from_numpy_dtype, + CircleEvaluationResult, + CircleReferenceEvaluator, + numpy_dtype_from_circle_tensor_type, +) +from test.support.circle.value_test import ( + CircleExtractionValueTestResult, + CirclePassValueTestResult, + CircleValueTestCase, + GraphInterfaceContract, + SignatureContract, + TensorContract, +) + +__all__ = [ + "CircleEvaluationResult", + "CircleExtractionValueTestResult", + "CircleModelBuilder", + "CirclePassValueTestResult", + "CircleReferenceEvaluator", + "CircleValueTestCase", + "GraphInterfaceContract", + "SignatureContract", + "TensorContract", + "circle_tensor_type_from_numpy_dtype", + "numpy_dtype_from_circle_tensor_type", +] diff --git a/test/support/circle/builder.py b/test/support/circle/builder.py new file mode 100644 index 00000000..a57ef1f4 --- /dev/null +++ b/test/support/circle/builder.py @@ -0,0 +1,473 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +import numpy as np +from circle_schema import circle + +from tico.circle.document import CircleDocument + +from test.support.circle.evaluator import ( + circle_tensor_type_from_numpy_dtype, + numpy_dtype_from_circle_tensor_type, +) + + +class CircleModelBuilder: + """Build small, executable Circle Object API fixtures for value tests.""" + + def __init__( + self, + *, + description: str = "circle-value-test", + subgraph_name: str = "main", + ) -> None: + """Create an empty single-subgraph Circle model.""" + + self.model = circle.Model.ModelT() + self.model.version = 0 + self.model.description = description + self.model.operatorCodes = [] + self.model.buffers = [circle.Buffer.BufferT()] + self.model.metadataBuffer = [] + self.model.metadata = [] + self.model.signatureDefs = [] + + self.subgraph = circle.SubGraph.SubGraphT() + self.subgraph.name = subgraph_name + self.subgraph.tensors = [] + self.subgraph.inputs = [] + self.subgraph.outputs = [] + self.subgraph.operators = [] + self.model.subgraphs = [self.subgraph] + + self._tensor_names: dict[str, int] = {} + self._operator_codes: dict[int, int] = {} + + def input( + self, + name: str, + shape: Sequence[int], + *, + dtype: np.dtype[Any] | type[Any] = np.float32, + shape_signature: Sequence[int] | None = None, + ) -> int: + """Add a graph input tensor and return its tensor index.""" + + tensor_index = self._add_tensor( + name, + shape, + dtype=dtype, + buffer_index=0, + shape_signature=shape_signature, + ) + self.subgraph.inputs.append(tensor_index) + return tensor_index + + def constant( + self, + name: str, + value: Any, + *, + dtype: np.dtype[Any] | type[Any] | None = None, + ) -> int: + """Add an inline constant tensor and return its tensor index.""" + + array = np.asarray(value, dtype=dtype) + array = np.ascontiguousarray(array) + circle_type = circle_tensor_type_from_numpy_dtype(array.dtype) + storage_dtype = numpy_dtype_from_circle_tensor_type(circle_type) + array = np.ascontiguousarray(array.astype(storage_dtype, copy=False)) + + buffer = circle.Buffer.BufferT() + buffer.data = array.reshape(-1).view(np.uint8) + self.model.buffers.append(buffer) + return self._add_tensor( + name, + array.shape, + dtype=array.dtype, + buffer_index=len(self.model.buffers) - 1, + ) + + def const_f32(self, name: str, value: Any) -> int: + """Add a FLOAT32 constant tensor.""" + + return self.constant(name, value, dtype=np.float32) + + def const_i32(self, name: str, value: Any) -> int: + """Add an INT32 constant tensor.""" + + return self.constant(name, value, dtype=np.int32) + + def add(self, lhs: int, rhs: int, *, name: str) -> int: + """Add an ADD operator and return its output tensor index.""" + + options = circle.AddOptions.AddOptionsT() + options.fusedActivationFunction = self._activation_none() + options.potScaleInt16 = False + return self._binary_operator( + self._builtin_operator("ADD"), + lhs, + rhs, + name=name, + options_type=self._builtin_options("AddOptions"), + options=options, + ) + + def sub(self, lhs: int, rhs: int, *, name: str) -> int: + """Add a SUB operator and return its output tensor index.""" + + options = circle.SubOptions.SubOptionsT() + options.fusedActivationFunction = self._activation_none() + options.potScaleInt16 = False + return self._binary_operator( + self._builtin_operator("SUB"), + lhs, + rhs, + name=name, + options_type=self._builtin_options("SubOptions"), + options=options, + ) + + def mul(self, lhs: int, rhs: int, *, name: str) -> int: + """Add a MUL operator and return its output tensor index.""" + + options = circle.MulOptions.MulOptionsT() + options.fusedActivationFunction = self._activation_none() + return self._binary_operator( + self._builtin_operator("MUL"), + lhs, + rhs, + name=name, + options_type=self._builtin_options("MulOptions"), + options=options, + ) + + def reshape( + self, + tensor_index: int, + new_shape: Sequence[int], + *, + name: str, + ) -> int: + """Add a RESHAPE operator with an inline INT32 shape tensor.""" + + input_tensor = self._tensor(tensor_index) + input_shape = tuple(int(value) for value in input_tensor.shape) + resolved_shape = self._resolve_reshape_shape(input_shape, new_shape) + shape_tensor_index = self.const_i32(f"{name}_shape", list(new_shape)) + output_index = self._add_tensor( + name, + resolved_shape, + dtype=numpy_dtype_from_circle_tensor_type(int(input_tensor.type)), + buffer_index=0, + ) + + options = circle.ReshapeOptions.ReshapeOptionsT() + options.newShape = [int(value) for value in new_shape] + self._append_operator( + self._builtin_operator("RESHAPE"), + inputs=[tensor_index, shape_tensor_index], + outputs=[output_index], + options_type=self._builtin_options("ReshapeOptions"), + options=options, + ) + return output_index + + def transpose( + self, + tensor_index: int, + permutation: Sequence[int], + *, + name: str, + ) -> int: + """Add a TRANSPOSE operator with an inline INT32 permutation tensor.""" + + input_tensor = self._tensor(tensor_index) + input_shape = tuple(int(value) for value in input_tensor.shape) + normalized_permutation = tuple(int(value) for value in permutation) + if len(normalized_permutation) != len(input_shape) or sorted( + normalized_permutation + ) != list(range(len(input_shape))): + raise ValueError( + f"Invalid permutation {normalized_permutation} for shape {input_shape}." + ) + + permutation_tensor_index = self.const_i32( + f"{name}_permutation", + normalized_permutation, + ) + output_shape = tuple(input_shape[axis] for axis in normalized_permutation) + output_index = self._add_tensor( + name, + output_shape, + dtype=numpy_dtype_from_circle_tensor_type(int(input_tensor.type)), + buffer_index=0, + ) + + options = circle.TransposeOptions.TransposeOptionsT() + self._append_operator( + self._builtin_operator("TRANSPOSE"), + inputs=[tensor_index, permutation_tensor_index], + outputs=[output_index], + options_type=self._builtin_options("TransposeOptions"), + options=options, + ) + return output_index + + def set_outputs(self, *tensor_indices: int) -> None: + """Set graph outputs in the supplied order.""" + + if not tensor_indices: + raise ValueError("At least one graph output is required.") + for tensor_index in tensor_indices: + self._tensor(tensor_index) + self.subgraph.outputs = [int(index) for index in tensor_indices] + + def tensor_index(self, name: str) -> int: + """Return the tensor index registered for a name.""" + + try: + return self._tensor_names[name] + except KeyError as error: + raise KeyError(f"Unknown Circle tensor name: {name!r}.") from error + + def operator_count(self) -> int: + """Return the number of operators currently in the subgraph.""" + + return len(self.subgraph.operators) + + def build( + self, + *, + signature_key: str | None = "serving_default", + ) -> CircleDocument: + """Finalize the fixture and return it as a CircleDocument.""" + + if not self.subgraph.outputs: + raise ValueError("Circle fixture does not define any graph outputs.") + if signature_key is None: + self.model.signatureDefs = [] + else: + self.model.signatureDefs = [self._make_signature(signature_key)] + return CircleDocument(self.model) + + def _add_tensor( + self, + name: str, + shape: Sequence[int], + *, + dtype: np.dtype[Any] | type[Any], + buffer_index: int, + shape_signature: Sequence[int] | None = None, + ) -> int: + """Create a tensor and return its subgraph-local index.""" + + if not name: + raise ValueError("Circle tensor names must not be empty.") + if name in self._tensor_names: + raise ValueError(f"Duplicate Circle tensor name: {name!r}.") + + normalized_shape = [int(dimension) for dimension in shape] + normalized_signature = ( + list(normalized_shape) + if shape_signature is None + else [int(dimension) for dimension in shape_signature] + ) + if len(normalized_shape) != len(normalized_signature): + raise ValueError("Shape and shape signature must have the same rank.") + + tensor = circle.Tensor.TensorT() + tensor.name = name + tensor.shape = normalized_shape + tensor.shapeSignature = normalized_signature + tensor.type = circle_tensor_type_from_numpy_dtype(dtype) + tensor.buffer = int(buffer_index) + tensor.isVariable = False + + tensor_index = len(self.subgraph.tensors) + self.subgraph.tensors.append(tensor) + self._tensor_names[name] = tensor_index + return tensor_index + + def _binary_operator( + self, + builtin_code: int, + lhs: int, + rhs: int, + *, + name: str, + options_type: int, + options: Any, + ) -> int: + """Add a broadcastable binary arithmetic operator.""" + + lhs_tensor = self._tensor(lhs) + rhs_tensor = self._tensor(rhs) + if int(lhs_tensor.type) != int(rhs_tensor.type): + raise TypeError( + f"Binary fixture inputs must have the same type: " + f"{lhs_tensor.type} != {rhs_tensor.type}." + ) + + output_shape = np.broadcast_shapes( + tuple(int(value) for value in lhs_tensor.shape), + tuple(int(value) for value in rhs_tensor.shape), + ) + output_index = self._add_tensor( + name, + output_shape, + dtype=numpy_dtype_from_circle_tensor_type(int(lhs_tensor.type)), + buffer_index=0, + ) + self._append_operator( + builtin_code, + inputs=[lhs, rhs], + outputs=[output_index], + options_type=options_type, + options=options, + ) + return output_index + + def _append_operator( + self, + builtin_code: int, + *, + inputs: Sequence[int], + outputs: Sequence[int], + options_type: int, + options: Any, + ) -> None: + """Append one builtin operator to the graph.""" + + operator = circle.Operator.OperatorT() + operator.opcodeIndex = self._operator_code_index(builtin_code) + operator.inputs = [int(index) for index in inputs] + operator.outputs = [int(index) for index in outputs] + operator.intermediates = [] + operator.mutatingVariableInputs = [] + operator.builtinOptionsType = int(options_type) + operator.builtinOptions = options + self.subgraph.operators.append(operator) + + def _operator_code_index(self, builtin_code: int) -> int: + """Return an existing operator-code index or create one.""" + + existing = self._operator_codes.get(int(builtin_code)) + if existing is not None: + return existing + + operator_code = circle.OperatorCode.OperatorCodeT() + operator_code.builtinCode = int(builtin_code) + operator_code.deprecatedBuiltinCode = min(127, int(builtin_code)) + operator_code.version = 1 + operator_code.customCode = None + + index = len(self.model.operatorCodes) + self.model.operatorCodes.append(operator_code) + self._operator_codes[int(builtin_code)] = index + return index + + def _make_signature(self, signature_key: str) -> Any: + """Create a signature matching the current graph interface.""" + + signature = circle.SignatureDef.SignatureDefT() + signature.signatureKey = signature_key + signature.subgraphIndex = 0 + signature.inputs = [ + self._make_tensor_map(self._tensor(index).name, index) + for index in self.subgraph.inputs + ] + signature.outputs = [ + self._make_tensor_map(self._tensor(index).name, index) + for index in self.subgraph.outputs + ] + return signature + + @staticmethod + def _make_tensor_map(name: str, tensor_index: int) -> Any: + """Create one signature tensor mapping.""" + + tensor_map = circle.TensorMap.TensorMapT() + tensor_map.name = name + tensor_map.tensorIndex = int(tensor_index) + return tensor_map + + def _tensor(self, tensor_index: int) -> Any: + """Return a tensor after validating its index.""" + + index = int(tensor_index) + if index < 0 or index >= len(self.subgraph.tensors): + raise IndexError( + f"Tensor index {index} is outside 0..{len(self.subgraph.tensors) - 1}." + ) + return self.subgraph.tensors[index] + + @staticmethod + def _resolve_reshape_shape( + input_shape: Sequence[int], + requested_shape: Sequence[int], + ) -> tuple[int, ...]: + """Resolve one optional inferred dimension in a reshape target.""" + + requested = [int(dimension) for dimension in requested_shape] + inferred_positions = [ + position for position, dimension in enumerate(requested) if dimension == -1 + ] + if len(inferred_positions) > 1: + raise ValueError("RESHAPE supports at most one inferred dimension.") + if any(dimension < -1 for dimension in requested): + raise ValueError(f"Invalid reshape target: {requested}.") + + input_elements = int(np.prod(tuple(input_shape), dtype=np.int64)) + known_elements = int( + np.prod( + [dimension for dimension in requested if dimension != -1], + dtype=np.int64, + ) + ) + if inferred_positions: + if known_elements == 0 or input_elements % known_elements != 0: + raise ValueError( + f"Cannot infer reshape target {requested} from shape {input_shape}." + ) + requested[inferred_positions[0]] = input_elements // known_elements + elif known_elements != input_elements: + raise ValueError( + f"Reshape target {requested} changes element count from " + f"{input_elements} to {known_elements}." + ) + return tuple(requested) + + @staticmethod + def _builtin_operator(name: str) -> int: + """Return a BuiltinOperator enum value.""" + + return int(getattr(circle.BuiltinOperator.BuiltinOperator, name)) + + @staticmethod + def _builtin_options(name: str) -> int: + """Return a BuiltinOptions enum value.""" + + return int(getattr(circle.BuiltinOptions.BuiltinOptions, name)) + + @staticmethod + def _activation_none() -> int: + """Return the NONE fused-activation enum value.""" + + return int(circle.ActivationFunctionType.ActivationFunctionType.NONE) diff --git a/test/support/circle/evaluator.py b/test/support/circle/evaluator.py new file mode 100644 index 00000000..2aca733b --- /dev/null +++ b/test/support/circle/evaluator.py @@ -0,0 +1,460 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable + +import numpy as np +from circle_schema import circle + +from tico.circle.document import CircleDocument +from tico.circle.graph import as_indices, as_list, OPTIONAL_TENSOR_INDEX + + +@dataclass(frozen=True) +class CircleEvaluationResult: + """Store graph outputs and all tensor values produced during evaluation.""" + + outputs: tuple[np.ndarray, ...] + tensor_values: dict[int, np.ndarray] + + +def _enum_value(enum_module_name: str, member_name: str) -> int: + """Return a generated Circle enum value with a descriptive failure.""" + + enum_module = getattr(circle, enum_module_name, None) + enum_type = ( + getattr(enum_module, enum_module_name, None) + if enum_module is not None + else None + ) + if enum_type is None or not hasattr(enum_type, member_name): + raise RuntimeError( + f"Circle schema does not provide {enum_module_name}.{member_name}." + ) + return int(getattr(enum_type, member_name)) + + +_TENSOR_TYPE_TO_DTYPE: dict[int, np.dtype[Any]] = { + _enum_value("TensorType", "FLOAT32"): np.dtype(" int: + """Return the Circle tensor type corresponding to a supported NumPy dtype.""" + + normalized = np.dtype(dtype).newbyteorder("=") + try: + return _DTYPE_TO_TENSOR_TYPE[normalized] + except KeyError as error: + raise NotImplementedError( + f"Circle value-test fixtures do not support NumPy dtype {normalized}." + ) from error + + +def _buffer_bytes(data: Any) -> bytes: + """Convert generated buffer payload data into immutable bytes.""" + + if isinstance(data, bytes): + return data + if isinstance(data, (bytearray, memoryview)): + return bytes(data) + if isinstance(data, np.ndarray): + return np.ascontiguousarray(data, dtype=np.uint8).tobytes() + try: + return bytes(data) + except (TypeError, ValueError) as error: + raise TypeError( + f"Unsupported Circle buffer payload type: {type(data).__name__}." + ) from error + + +def _shape_tuple(tensor: Any) -> tuple[int, ...]: + """Return a tensor shape as a tuple of Python integers.""" + + return tuple( + int(dimension) for dimension in as_list(getattr(tensor, "shape", None)) + ) + + +def _builtin_operator_code(operator_code: Any) -> int: + """Resolve the builtin code stored in an OperatorCode object.""" + + builtin_code = int(getattr(operator_code, "builtinCode", 0) or 0) + deprecated_code = int( + getattr(operator_code, "deprecatedBuiltinCode", builtin_code) or 0 + ) + + placeholder = getattr( + getattr(circle.BuiltinOperator, "BuiltinOperator", object()), + "PLACEHOLDER_FOR_GREATER_OP_CODES", + 127, + ) + if builtin_code == 0 and deprecated_code != 0: + return deprecated_code + if deprecated_code != int(placeholder) and builtin_code == int(placeholder): + return deprecated_code + return builtin_code + + +class CircleReferenceEvaluator: + """Evaluate a deliberately small subset of Circle operators with NumPy. + + The evaluator is test-only infrastructure. It is intentionally not a general + Circle runtime and rejects unsupported data types, operators, fused + activations, external buffers, and optional inputs instead of guessing their + semantics. + """ + + def __init__(self) -> None: + """Create an evaluator with handlers for value-test operators.""" + + self._handlers: dict[ + int, + Callable[[Any, tuple[np.ndarray, ...]], tuple[np.ndarray, ...]], + ] = { + _enum_value("BuiltinOperator", "ADD"): self._evaluate_add, + _enum_value("BuiltinOperator", "SUB"): self._evaluate_sub, + _enum_value("BuiltinOperator", "MUL"): self._evaluate_mul, + _enum_value("BuiltinOperator", "RESHAPE"): self._evaluate_reshape, + _enum_value("BuiltinOperator", "TRANSPOSE"): self._evaluate_transpose, + } + + def evaluate( + self, + document: CircleDocument, + inputs: tuple[np.ndarray, ...], + *, + subgraph_index: int = 0, + ) -> CircleEvaluationResult: + """Evaluate one subgraph and return outputs plus intermediate values.""" + + subgraph = document.subgraph(subgraph_index) + tensors = as_list(getattr(subgraph, "tensors", None)) + operators = as_list(getattr(subgraph, "operators", None)) + input_indices = as_indices(getattr(subgraph, "inputs", None)) + output_indices = as_indices(getattr(subgraph, "outputs", None)) + + if len(inputs) != len(input_indices): + raise ValueError( + "Circle input count mismatch: " + f"expected {len(input_indices)}, received {len(inputs)}." + ) + + tensor_values: dict[int, np.ndarray] = {} + for tensor_index, input_value in zip(input_indices, inputs): + tensor_values[tensor_index] = self._validate_tensor_value( + tensors[tensor_index], + np.asarray(input_value), + path=f"subgraphs[{subgraph_index}].inputs[{tensor_index}]", + ) + + for tensor_index, tensor in enumerate(tensors): + if tensor_index in tensor_values: + continue + constant = self._decode_constant(document, tensor, tensor_index) + if constant is not None: + tensor_values[tensor_index] = constant + + operator_codes = as_list(getattr(document.model, "operatorCodes", None)) + for operator_index, operator in enumerate(operators): + opcode_index = int(getattr(operator, "opcodeIndex", -1)) + if opcode_index < 0 or opcode_index >= len(operator_codes): + raise ValueError( + f"Operator {operator_index} references invalid opcode index " + f"{opcode_index}." + ) + + builtin_code = _builtin_operator_code(operator_codes[opcode_index]) + try: + handler = self._handlers[builtin_code] + except KeyError as error: + raise NotImplementedError( + "CircleReferenceEvaluator does not support builtin operator " + f"{builtin_code} at operator {operator_index}." + ) from error + + operator_inputs = self._resolve_operator_inputs( + operator, + tensor_values, + operator_index=operator_index, + ) + output_values = handler(operator, operator_inputs) + output_indices_for_operator = as_indices(getattr(operator, "outputs", None)) + if len(output_values) != len(output_indices_for_operator): + raise ValueError( + f"Operator {operator_index} produced {len(output_values)} values " + f"for {len(output_indices_for_operator)} output tensors." + ) + + for tensor_index, output_value in zip( + output_indices_for_operator, output_values + ): + if tensor_index < 0 or tensor_index >= len(tensors): + raise ValueError( + f"Operator {operator_index} references invalid output tensor " + f"{tensor_index}." + ) + tensor_values[tensor_index] = self._validate_tensor_value( + tensors[tensor_index], + np.asarray(output_value), + path=( + f"subgraphs[{subgraph_index}].operators[{operator_index}]" + f".outputs[{tensor_index}]" + ), + ) + + missing_outputs = [ + tensor_index + for tensor_index in output_indices + if tensor_index not in tensor_values + ] + if missing_outputs: + raise ValueError( + f"Circle graph outputs were not evaluated: {missing_outputs}." + ) + + outputs = tuple( + np.array(tensor_values[tensor_index], copy=True) + for tensor_index in output_indices + ) + copied_values = { + tensor_index: np.array(value, copy=True) + for tensor_index, value in tensor_values.items() + } + return CircleEvaluationResult(outputs=outputs, tensor_values=copied_values) + + def _decode_constant( + self, + document: CircleDocument, + tensor: Any, + tensor_index: int, + ) -> np.ndarray | None: + """Decode one inline constant tensor or return None for activations.""" + + if bool(getattr(tensor, "isVariable", False)): + return None + + buffer_index = int(getattr(tensor, "buffer", 0) or 0) + if buffer_index == 0: + return None + + buffers = as_list(getattr(document.model, "buffers", None)) + if buffer_index < 0 or buffer_index >= len(buffers): + raise ValueError( + f"Tensor {tensor_index} references invalid buffer {buffer_index}." + ) + + buffer = buffers[buffer_index] + offset = int(getattr(buffer, "offset", 0) or 0) + size = int(getattr(buffer, "size", 0) or 0) + if offset or size: + raise NotImplementedError( + "CircleReferenceEvaluator does not support external buffers." + ) + + data = getattr(buffer, "data", None) + if data is None: + return None + raw = _buffer_bytes(data) + if not raw: + return None + + dtype = numpy_dtype_from_circle_tensor_type(int(getattr(tensor, "type"))) + shape = _shape_tuple(tensor) + element_count = int(np.prod(shape, dtype=np.int64)) if shape else 1 + expected_size = element_count * dtype.itemsize + if len(raw) != expected_size: + raise ValueError( + f"Tensor {tensor_index} expects {expected_size} buffer bytes for " + f"shape {shape} and dtype {dtype}, but buffer {buffer_index} " + f"contains {len(raw)} bytes." + ) + + value = np.frombuffer(raw, dtype=dtype, count=element_count).copy() + return value.reshape(shape if shape else ()) + + def _resolve_operator_inputs( + self, + operator: Any, + tensor_values: dict[int, np.ndarray], + *, + operator_index: int, + ) -> tuple[np.ndarray, ...]: + """Resolve operator inputs from the evaluated tensor map.""" + + values: list[np.ndarray] = [] + for position, tensor_index in enumerate( + as_indices(getattr(operator, "inputs", None)) + ): + if tensor_index == OPTIONAL_TENSOR_INDEX: + raise NotImplementedError( + "CircleReferenceEvaluator does not support optional inputs." + ) + try: + values.append(tensor_values[tensor_index]) + except KeyError as error: + raise ValueError( + f"Operator {operator_index} input {position} references tensor " + f"{tensor_index}, which has not been evaluated." + ) from error + return tuple(values) + + def _validate_tensor_value( + self, + tensor: Any, + value: np.ndarray, + *, + path: str, + ) -> np.ndarray: + """Validate and copy a value against a Circle tensor contract.""" + + expected_shape = _shape_tuple(tensor) + if tuple(value.shape) != expected_shape: + raise ValueError( + f"{path} has shape {tuple(value.shape)}, expected {expected_shape}." + ) + + expected_dtype = numpy_dtype_from_circle_tensor_type( + int(getattr(tensor, "type")) + ).newbyteorder("=") + actual_dtype = value.dtype.newbyteorder("=") + if actual_dtype != expected_dtype: + raise TypeError( + f"{path} has dtype {actual_dtype}, expected {expected_dtype}." + ) + return np.array(value, copy=True, order="C") + + def _require_no_fused_activation(self, operator: Any) -> None: + """Reject arithmetic operators with a non-NONE fused activation.""" + + options = getattr(operator, "builtinOptions", None) + activation = int(getattr(options, "fusedActivationFunction", 0) or 0) + none_value = _enum_value("ActivationFunctionType", "NONE") + if activation != none_value: + raise NotImplementedError( + "CircleReferenceEvaluator supports only fused activation NONE." + ) + + def _evaluate_add( + self, + operator: Any, + inputs: tuple[np.ndarray, ...], + ) -> tuple[np.ndarray, ...]: + """Evaluate an ADD operator.""" + + self._require_no_fused_activation(operator) + self._require_input_count("ADD", inputs, 2) + return (np.add(inputs[0], inputs[1]),) + + def _evaluate_sub( + self, + operator: Any, + inputs: tuple[np.ndarray, ...], + ) -> tuple[np.ndarray, ...]: + """Evaluate a SUB operator.""" + + self._require_no_fused_activation(operator) + self._require_input_count("SUB", inputs, 2) + return (np.subtract(inputs[0], inputs[1]),) + + def _evaluate_mul( + self, + operator: Any, + inputs: tuple[np.ndarray, ...], + ) -> tuple[np.ndarray, ...]: + """Evaluate a MUL operator.""" + + self._require_no_fused_activation(operator) + self._require_input_count("MUL", inputs, 2) + return (np.multiply(inputs[0], inputs[1]),) + + def _evaluate_reshape( + self, + operator: Any, + inputs: tuple[np.ndarray, ...], + ) -> tuple[np.ndarray, ...]: + """Evaluate a RESHAPE operator.""" + + if len(inputs) == 2: + target_shape = tuple(int(value) for value in inputs[1].reshape(-1)) + elif len(inputs) == 1: + options = getattr(operator, "builtinOptions", None) + target_shape = tuple( + int(value) for value in as_list(getattr(options, "newShape", None)) + ) + else: + raise ValueError( + f"RESHAPE expects one or two inputs, received {len(inputs)}." + ) + return (np.reshape(inputs[0], target_shape),) + + def _evaluate_transpose( + self, + operator: Any, + inputs: tuple[np.ndarray, ...], + ) -> tuple[np.ndarray, ...]: + """Evaluate a TRANSPOSE operator.""" + + del operator + self._require_input_count("TRANSPOSE", inputs, 2) + permutation = tuple(int(value) for value in inputs[1].reshape(-1)) + rank = inputs[0].ndim + if len(permutation) != rank or sorted(permutation) != list(range(rank)): + raise ValueError( + f"Invalid TRANSPOSE permutation {permutation} for rank {rank}." + ) + return (np.transpose(inputs[0], axes=permutation),) + + @staticmethod + def _require_input_count( + operator_name: str, + inputs: tuple[np.ndarray, ...], + expected: int, + ) -> None: + """Require an exact operator input count.""" + + if len(inputs) != expected: + raise ValueError( + f"{operator_name} expects {expected} inputs, received {len(inputs)}." + ) diff --git a/test/support/circle/value_test.py b/test/support/circle/value_test.py new file mode 100644 index 00000000..ecff2b4f --- /dev/null +++ b/test/support/circle/value_test.py @@ -0,0 +1,534 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import unittest +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from typing import Any + +import numpy as np + +from tico.circle._schema import decode_text +from tico.circle.document import CircleDocument +from tico.circle.graph import as_indices, as_list + +from test.support.circle.evaluator import ( + CircleEvaluationResult, + CircleReferenceEvaluator, +) + + +@dataclass(frozen=True) +class TensorContract: + """Describe the semantically relevant interface fields of one tensor.""" + + name: str + shape: tuple[int, ...] + shape_signature: tuple[int, ...] | None + tensor_type: int + is_variable: bool + quantization: Any + + +@dataclass(frozen=True) +class SignatureContract: + """Describe one signature without relying on mutable tensor indices.""" + + key: str + inputs: tuple[tuple[str, TensorContract], ...] + outputs: tuple[tuple[str, TensorContract], ...] + + +@dataclass(frozen=True) +class GraphInterfaceContract: + """Describe graph input, output, and signature contracts.""" + + inputs: tuple[TensorContract, ...] + outputs: tuple[TensorContract, ...] + signatures: tuple[SignatureContract, ...] + + +@dataclass(frozen=True) +class CirclePassValueTestResult: + """Return transformed artifacts and numerical results from a pass value test.""" + + document: CircleDocument + transform_result: Any + source_evaluation: CircleEvaluationResult + transformed_evaluation: CircleEvaluationResult + + +@dataclass(frozen=True) +class CircleExtractionValueTestResult: + """Return extraction metadata and numerical results from a value test.""" + + extraction_result: Any + document: CircleDocument + source_evaluation: CircleEvaluationResult + extracted_evaluation: CircleEvaluationResult + + @property + def selected_operator_indices(self) -> tuple[int, ...]: + """Return the source operator indices selected for extraction.""" + + return self.extraction_result.selected_operator_indices + + @property + def source_boundary(self) -> Any: + """Return the extraction boundary in the source tensor index space.""" + + return self.extraction_result.source_boundary + + @property + def boundary(self) -> Any: + """Return the extraction boundary in the compacted tensor index space.""" + + return self.extraction_result.boundary + + @property + def removed_operators(self) -> int: + """Return the number of operators removed by extraction.""" + + return int(self.extraction_result.removed_operators) + + @property + def rewrite_stats(self) -> Any: + """Return the source extraction rewrite statistics.""" + + return self.extraction_result.rewrite_stats + + +def _freeze_object(value: Any) -> Any: + """Convert generated Object API values into recursively comparable data.""" + + if value is None or isinstance(value, (bool, int, float, str, bytes)): + return value + if isinstance(value, np.ndarray): + return (str(value.dtype), tuple(value.shape), value.tobytes()) + if isinstance(value, (list, tuple)): + return tuple(_freeze_object(item) for item in value) + attributes = getattr(value, "__dict__", None) + if attributes is not None: + return tuple( + sorted( + (name, _freeze_object(attribute)) + for name, attribute in attributes.items() + ) + ) + return repr(value) + + +class CircleValueTestCase(unittest.TestCase): + """Provide reusable numerical assertions for Circle graph transformations.""" + + evaluator: CircleReferenceEvaluator + + def setUp(self) -> None: + """Create a fresh reference evaluator for each test.""" + + super().setUp() + self.evaluator = CircleReferenceEvaluator() + + def round_trip(self, document: CircleDocument) -> CircleDocument: + """Serialize, deserialize, and structurally verify a Circle document.""" + + restored = CircleDocument.from_bytes(document.to_bytes()) + restored.verify(raise_on_error=True) + return restored + + def assert_outputs_equal( + self, + expected: Sequence[np.ndarray], + actual: Sequence[np.ndarray], + *, + rtol: float = 0.0, + atol: float = 0.0, + ) -> None: + """Compare output count, shape, dtype, and numerical values.""" + + self.assertEqual(len(expected), len(actual)) + for output_index, (expected_value, actual_value) in enumerate( + zip(expected, actual) + ): + expected_array = np.asarray(expected_value) + actual_array = np.asarray(actual_value) + self.assertEqual( + expected_array.shape, + actual_array.shape, + msg=f"Output {output_index} shape mismatch.", + ) + self.assertEqual( + expected_array.dtype, + actual_array.dtype, + msg=f"Output {output_index} dtype mismatch.", + ) + if np.issubdtype(expected_array.dtype, np.inexact): + np.testing.assert_allclose( + actual_array, + expected_array, + rtol=rtol, + atol=atol, + equal_nan=True, + err_msg=f"Output {output_index} value mismatch.", + ) + else: + np.testing.assert_array_equal( + actual_array, + expected_array, + err_msg=f"Output {output_index} value mismatch.", + ) + + def tensor_contract( + self, + document: CircleDocument, + tensor_index: int, + *, + subgraph_index: int = 0, + ) -> TensorContract: + """Return a stable tensor contract independent of index compaction.""" + + tensor = document.subgraph(subgraph_index).tensors[int(tensor_index)] + raw_shape_signature = getattr(tensor, "shapeSignature", None) + shape_signature = ( + None + if raw_shape_signature is None + else tuple(int(value) for value in as_list(raw_shape_signature)) + ) + return TensorContract( + name=decode_text(getattr(tensor, "name", "")), + shape=tuple( + int(value) for value in as_list(getattr(tensor, "shape", None)) + ), + shape_signature=shape_signature, + tensor_type=int(getattr(tensor, "type")), + is_variable=bool(getattr(tensor, "isVariable", False)), + quantization=_freeze_object(getattr(tensor, "quantization", None)), + ) + + def graph_interface_contract( + self, + document: CircleDocument, + *, + subgraph_index: int = 0, + ) -> GraphInterfaceContract: + """Return graph interface semantics without using raw tensor indices.""" + + subgraph = document.subgraph(subgraph_index) + inputs = tuple( + self.tensor_contract( + document, + tensor_index, + subgraph_index=subgraph_index, + ) + for tensor_index in as_indices(getattr(subgraph, "inputs", None)) + ) + outputs = tuple( + self.tensor_contract( + document, + tensor_index, + subgraph_index=subgraph_index, + ) + for tensor_index in as_indices(getattr(subgraph, "outputs", None)) + ) + + signatures: list[SignatureContract] = [] + for signature in as_list(getattr(document.model, "signatureDefs", None)): + if int(getattr(signature, "subgraphIndex", -1)) != subgraph_index: + continue + signature_inputs = tuple( + ( + decode_text(getattr(tensor_map, "name", "")), + self.tensor_contract( + document, + int(getattr(tensor_map, "tensorIndex")), + subgraph_index=subgraph_index, + ), + ) + for tensor_map in as_list(getattr(signature, "inputs", None)) + ) + signature_outputs = tuple( + ( + decode_text(getattr(tensor_map, "name", "")), + self.tensor_contract( + document, + int(getattr(tensor_map, "tensorIndex")), + subgraph_index=subgraph_index, + ), + ) + for tensor_map in as_list(getattr(signature, "outputs", None)) + ) + signatures.append( + SignatureContract( + key=decode_text(getattr(signature, "signatureKey", "")), + inputs=signature_inputs, + outputs=signature_outputs, + ) + ) + + return GraphInterfaceContract( + inputs=inputs, + outputs=outputs, + signatures=tuple(signatures), + ) + + def assert_interfaces_equal( + self, + expected_document: CircleDocument, + actual_document: CircleDocument, + *, + subgraph_index: int = 0, + check_tensor_names: bool = False, + ) -> None: + """Assert that two documents expose equivalent graph interfaces.""" + + expected = self.graph_interface_contract( + expected_document, + subgraph_index=subgraph_index, + ) + actual = self.graph_interface_contract( + actual_document, + subgraph_index=subgraph_index, + ) + self.assertEqual(len(expected.inputs), len(actual.inputs)) + self.assertEqual(len(expected.outputs), len(actual.outputs)) + + for expected_tensor, actual_tensor in zip(expected.inputs, actual.inputs): + self._assert_tensor_contracts_equal( + expected_tensor, + actual_tensor, + check_name=check_tensor_names, + ) + for expected_tensor, actual_tensor in zip(expected.outputs, actual.outputs): + self._assert_tensor_contracts_equal( + expected_tensor, + actual_tensor, + check_name=check_tensor_names, + ) + + self.assertEqual(len(expected.signatures), len(actual.signatures)) + for expected_signature, actual_signature in zip( + expected.signatures, + actual.signatures, + ): + self.assertEqual(expected_signature.key, actual_signature.key) + self.assertEqual( + [name for name, _ in expected_signature.inputs], + [name for name, _ in actual_signature.inputs], + ) + self.assertEqual( + [name for name, _ in expected_signature.outputs], + [name for name, _ in actual_signature.outputs], + ) + for (_, expected_tensor), (_, actual_tensor) in zip( + expected_signature.inputs, + actual_signature.inputs, + ): + self._assert_tensor_contracts_equal( + expected_tensor, + actual_tensor, + check_name=check_tensor_names, + ) + for (_, expected_tensor), (_, actual_tensor) in zip( + expected_signature.outputs, + actual_signature.outputs, + ): + self._assert_tensor_contracts_equal( + expected_tensor, + actual_tensor, + check_name=check_tensor_names, + ) + + def assert_tensor_contract_equal( + self, + expected_document: CircleDocument, + expected_tensor_index: int, + actual_document: CircleDocument, + actual_tensor_index: int, + *, + expected_subgraph_index: int = 0, + actual_subgraph_index: int = 0, + check_name: bool = True, + ) -> None: + """Compare two tensor contracts across rewritten index spaces.""" + + self._assert_tensor_contracts_equal( + self.tensor_contract( + expected_document, + expected_tensor_index, + subgraph_index=expected_subgraph_index, + ), + self.tensor_contract( + actual_document, + actual_tensor_index, + subgraph_index=actual_subgraph_index, + ), + check_name=check_name, + ) + + def assert_pass_preserves_value( + self, + source: CircleDocument, + inputs: tuple[np.ndarray, ...], + transform: Callable[[CircleDocument], Any], + *, + expected_outputs: Sequence[np.ndarray], + expected_modified: bool = True, + expected_changes: int | None = None, + compare_interface: bool = True, + check_interface_tensor_names: bool = False, + rtol: float = 0.0, + atol: float = 0.0, + ) -> CirclePassValueTestResult: + """Assert that a Circle transformation preserves a numerical golden.""" + + source = self.round_trip(source) + source_evaluation = self.evaluator.evaluate(source, inputs) + self.assert_outputs_equal( + expected_outputs, + source_evaluation.outputs, + rtol=rtol, + atol=atol, + ) + + transformed = source.clone() + transform_result = transform(transformed) + if hasattr(transform_result, "modified"): + self.assertEqual(bool(transform_result.modified), expected_modified) + elif expected_modified: + self.fail("Transformation result does not expose a modified property.") + if expected_changes is not None: + self.assertEqual(int(transform_result.changes), expected_changes) + + transformed = self.round_trip(transformed) + if compare_interface: + self.assert_interfaces_equal( + source, + transformed, + check_tensor_names=check_interface_tensor_names, + ) + + transformed_evaluation = self.evaluator.evaluate(transformed, inputs) + self.assert_outputs_equal( + expected_outputs, + transformed_evaluation.outputs, + rtol=rtol, + atol=atol, + ) + self.assert_outputs_equal( + source_evaluation.outputs, + transformed_evaluation.outputs, + rtol=rtol, + atol=atol, + ) + return CirclePassValueTestResult( + document=transformed, + transform_result=transform_result, + source_evaluation=source_evaluation, + transformed_evaluation=transformed_evaluation, + ) + + def assert_extraction_preserves_value( + self, + source: CircleDocument, + inputs: tuple[np.ndarray, ...], + extract: Callable[[CircleDocument], Any], + *, + expected_source_outputs: Sequence[np.ndarray] | None = None, + rtol: float = 0.0, + atol: float = 0.0, + ) -> CircleExtractionValueTestResult: + """Assert that an extracted graph reproduces its source boundary values.""" + + source = self.round_trip(source) + source_evaluation = self.evaluator.evaluate(source, inputs) + if expected_source_outputs is not None: + self.assert_outputs_equal( + expected_source_outputs, + source_evaluation.outputs, + rtol=rtol, + atol=atol, + ) + + extraction_result = extract(source) + extracted = self.round_trip(extraction_result.document) + extracted_inputs = tuple( + source_evaluation.tensor_values[tensor_index] + for tensor_index in extraction_result.source_boundary.inputs + ) + expected_outputs = tuple( + source_evaluation.tensor_values[tensor_index] + for tensor_index in extraction_result.source_boundary.outputs + ) + extracted_evaluation = self.evaluator.evaluate(extracted, extracted_inputs) + self.assert_outputs_equal( + expected_outputs, + extracted_evaluation.outputs, + rtol=rtol, + atol=atol, + ) + + self.assertEqual( + len(extraction_result.source_boundary.inputs), + len(extraction_result.boundary.inputs), + ) + self.assertEqual( + len(extraction_result.source_boundary.outputs), + len(extraction_result.boundary.outputs), + ) + for source_index, extracted_index in zip( + extraction_result.source_boundary.inputs, + extraction_result.boundary.inputs, + ): + self.assert_tensor_contract_equal( + source, + source_index, + extracted, + extracted_index, + ) + for source_index, extracted_index in zip( + extraction_result.source_boundary.outputs, + extraction_result.boundary.outputs, + ): + self.assert_tensor_contract_equal( + source, + source_index, + extracted, + extracted_index, + ) + + return CircleExtractionValueTestResult( + extraction_result=extraction_result, + document=extracted, + source_evaluation=source_evaluation, + extracted_evaluation=extracted_evaluation, + ) + + def _assert_tensor_contracts_equal( + self, + expected: TensorContract, + actual: TensorContract, + *, + check_name: bool, + ) -> None: + """Compare tensor contract fields with optional name checking.""" + + if check_name: + self.assertEqual(expected.name, actual.name) + self.assertEqual(expected.shape, actual.shape) + self.assertEqual(expected.shape_signature, actual.shape_signature) + self.assertEqual(expected.tensor_type, actual.tensor_type) + self.assertEqual(expected.is_variable, actual.is_variable) + self.assertEqual(expected.quantization, actual.quantization) diff --git a/test/unit_test/circle/passes/optimization/remove/test_layout_ops.py b/test/unit_test/circle/passes/optimization/remove/test_layout_ops.py index c2ec7db7..1d1ab132 100644 --- a/test/unit_test/circle/passes/optimization/remove/test_layout_ops.py +++ b/test/unit_test/circle/passes/optimization/remove/test_layout_ops.py @@ -24,6 +24,8 @@ _get_const_data, _is_reshape_op, _is_transpose_op, + _RESHAPE_BUILTIN_CODE, + _TRANSPOSE_BUILTIN_CODE, RemoveRedundantLayoutOpsPass, ) @@ -41,6 +43,7 @@ def _make_inverse_transpose_document() -> CircleDocument: """Create a non-square graph containing two inverse transposes.""" + permutation = struct.pack(" CircleDocument: FakeBuffer(data=permutation), FakeBuffer(data=permutation), ], - operatorCodes=[FakeOperatorCode(builtinCode=54)], + operatorCodes=[FakeOperatorCode(builtinCode=_TRANSPOSE_BUILTIN_CODE)], signatureDefs=[ FakeSignatureDef( signatureKey="main", @@ -83,29 +86,31 @@ class TestPermutationHelpers(unittest.TestCase): def test_check_perm_identity(self): """Test identity permutation.""" - # perm = [0, 1, 2] + result = _check_perm([0, 1, 2], [0, 1, 2]) self.assertTrue(result) def test_check_perm_inverse_2d(self): - """Test 2D inverse permutation (swap).""" - # perm1 = [1, 0], perm2 = [1, 0] + """Test a self-inverse 2D permutation.""" + result = _check_perm([1, 0], [1, 0]) self.assertTrue(result) def test_check_perm_inverse_3d(self): - """Test 3D inverse permutation.""" - # perm1 = [0, 2, 1], perm2 = [0, 2, 1] - result = _check_perm([0, 2, 1], [0, 2, 1]) + """Test two non-self-inverse 3D permutations.""" + + result = _check_perm([2, 0, 1], [1, 2, 0]) self.assertTrue(result) def test_check_perm_non_inverse(self): """Test non-inverse permutations.""" + result = _check_perm([1, 0, 2], [0, 2, 1]) self.assertFalse(result) def test_check_perm_length_mismatch(self): """Test permutations of different lengths.""" + result = _check_perm([1, 0], [0, 2, 1]) self.assertFalse(result) @@ -115,28 +120,31 @@ class TestOperatorDetection(unittest.TestCase): def test_is_reshape_valid(self): """Test valid Reshape operator detection.""" + operator = Mock() operator.opcodeIndex = 0 opcode = Mock() - opcode.builtinCode = 26 # RESHAPE + opcode.builtinCode = _RESHAPE_BUILTIN_CODE result = _is_reshape_op(operator, [opcode]) self.assertTrue(result) def test_is_reshape_invalid_type(self): - """Test non-Reshape operator.""" + """Test non-Reshape operator detection.""" + operator = Mock() operator.opcodeIndex = 0 opcode = Mock() - opcode.builtinCode = 1 # ADD + opcode.builtinCode = 0 # ADD result = _is_reshape_op(operator, [opcode]) self.assertFalse(result) def test_is_reshape_invalid_index(self): """Test Reshape with invalid opcode index.""" + operator = Mock() operator.opcodeIndex = 5 @@ -145,22 +153,24 @@ def test_is_reshape_invalid_index(self): def test_is_transpose_valid(self): """Test valid Transpose operator detection.""" + operator = Mock() operator.opcodeIndex = 0 opcode = Mock() - opcode.builtinCode = 54 # TRANSPOSE + opcode.builtinCode = _TRANSPOSE_BUILTIN_CODE result = _is_transpose_op(operator, [opcode]) self.assertTrue(result) def test_is_transpose_invalid_type(self): - """Test non-Transpose operator.""" + """Test non-Transpose operator detection.""" + operator = Mock() operator.opcodeIndex = 0 opcode = Mock() - opcode.builtinCode = 26 # RESHAPE + opcode.builtinCode = _RESHAPE_BUILTIN_CODE result = _is_transpose_op(operator, [opcode]) self.assertFalse(result) @@ -171,6 +181,7 @@ class TestConstDataExtraction(unittest.TestCase): def test_get_const_data_valid(self): """Test extracting valid int32 constant data.""" + graph = Mock() tensor = Mock() @@ -190,7 +201,8 @@ def test_get_const_data_valid(self): self.assertEqual(result, [1, 0]) def test_get_const_data_invalid_tensor_index(self): - """Test with invalid tensor index.""" + """Test constant extraction with an invalid tensor index.""" + graph = Mock() graph.subgraph = Mock() graph.subgraph.tensors = [] @@ -199,7 +211,8 @@ def test_get_const_data_invalid_tensor_index(self): self.assertIsNone(result) def test_get_const_data_no_buffer(self): - """Test with missing buffer data.""" + """Test constant extraction without buffer data.""" + graph = Mock() tensor = Mock() @@ -221,21 +234,21 @@ class TestRemoveRedundantLayoutOpsPass(unittest.TestCase): def test_pass_name(self): """Test pass name property.""" + pass_obj = RemoveRedundantLayoutOpsPass() self.assertEqual(pass_obj.name, "RemoveRedundantLayoutOpsPass") def test_run_with_mock_document(self): - """Test pass run method with minimal mock.""" + """Test pass execution with a minimal mock document.""" + document = Mock() document.subgraph_count = 1 graph = Mock() graph.subgraph = Mock() - # Make operators iterable for as_list() operators: list[object] = [] graph.subgraph.operators = operators - # Make operatorCodes iterable operator_codes: list[object] = [] model = Mock() model.operatorCodes = operator_codes @@ -253,27 +266,25 @@ def test_run_with_mock_document(self): self.assertEqual(result.changes, 0) def test_remove_redundant_reshape_simple(self): - """Test removing redundant Reshape operation.""" + """Test bypassing the first of two consecutive Reshape operators.""" + pass_obj = RemoveRedundantLayoutOpsPass() graph = Mock() - - # reshape1 is a Reshape operation reshape1 = Mock() - reshape1.inputs = [0, 1] # input=0, shape=1 + reshape1.inputs = [0, 1] reshape1.opcodeIndex = 0 - # reshape2 is also a Reshape operation that takes reshape1's output as input reshape2 = Mock() - reshape2.inputs = [2, 3] # input=2 (reshape1's output), shape=3 + reshape2.inputs = [2, 3] reshape2.opcodeIndex = 0 graph.subgraph = Mock() graph.subgraph.operators = [reshape1, reshape2] - graph.producer = Mock(return_value=0) # reshape1 produces tensor 2 + graph.producer = Mock(return_value=0) opcode = Mock() - opcode.builtinCode = 26 # RESHAPE + opcode.builtinCode = _RESHAPE_BUILTIN_CODE graph.model = Mock() graph.model.operatorCodes = [opcode] @@ -284,7 +295,8 @@ def test_remove_redundant_reshape_simple(self): self.assertEqual(reshape2.inputs[0], 0) def test_remove_redundant_reshape_no_producer(self): - """Test Reshape with no producer (graph input).""" + """Test a Reshape whose input is a graph input.""" + pass_obj = RemoveRedundantLayoutOpsPass() graph = Mock() @@ -298,6 +310,7 @@ def test_remove_redundant_reshape_no_producer(self): def test_remove_redundant_transpose_inverse(self): """Test inverse Transpose cancellation and dead-code removal.""" + document = _make_inverse_transpose_document() subgraph = document.subgraph() second_transpose = subgraph.operators[1] @@ -324,6 +337,7 @@ def test_remove_redundant_transpose_inverse(self): def test_inverse_transpose_preserves_shared_first_transpose(self): """Test that a shared first Transpose remains live after cancellation.""" + document = _make_inverse_transpose_document() subgraph = document.subgraph() first_transpose = subgraph.operators[0] @@ -348,16 +362,17 @@ def test_inverse_transpose_preserves_shared_first_transpose(self): self.assertTrue(document.verify(raise_on_error=False).ok) def test_remove_redundant_transpose_missing_perm(self): - """Test Transpose with missing permutation data.""" + """Test Transpose handling when permutation data is missing.""" + pass_obj = RemoveRedundantLayoutOpsPass() graph = Mock() graph.subgraph = Mock() - graph.subgraph.tensors = [None] # Minimal tensors list + graph.subgraph.tensors = [None] graph.subgraph.operators = [Mock()] graph.model = Mock() - graph.model.buffers = [None] # Minimal buffers list + graph.model.buffers = [None] graph.model.operatorCodes = [] transpose = Mock() diff --git a/test/unit_test/circle/value/__init__.py b/test/unit_test/circle/value/__init__.py new file mode 100644 index 00000000..f6769d58 --- /dev/null +++ b/test/unit_test/circle/value/__init__.py @@ -0,0 +1,15 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Execution-based value tests for Circle artifact transformations.""" diff --git a/test/unit_test/circle/value/test_cleanup_value.py b/test/unit_test/circle/value/test_cleanup_value.py new file mode 100644 index 00000000..9b9f35f2 --- /dev/null +++ b/test/unit_test/circle/value/test_cleanup_value.py @@ -0,0 +1,84 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import numpy as np + +from tico.circle._schema import decode_text +from tico.circle.passes import CirclePassContext, CirclePassManager +from tico.circle.passes.cleanup import CompactIndicesPass, DeadCodeEliminationPass + +from test.support.circle.builder import CircleModelBuilder +from test.support.circle.value_test import CircleValueTestCase + + +class CircleCleanupValueTest(CircleValueTestCase): + """Check value preservation across dead-code elimination and compaction.""" + + def test_dce_and_compaction_preserve_values_with_interleaved_dead_objects(self): + """Remove an interleaved dead branch without changing live outputs.""" + + builder = CircleModelBuilder(description="cleanup-value-test") + x = builder.input("x", [3]) + live_addend = builder.const_f32("live_addend", 2.0) + added = builder.add(x, live_addend, name="added") + + dead_addend = builder.const_f32("dead_addend", 100.0) + builder.sub(x, dead_addend, name="dead_output") + + live_multiplier = builder.const_f32("live_multiplier", 3.0) + output = builder.mul(added, live_multiplier, name="output") + builder.set_outputs(output) + source = builder.build() + self.assertEqual(len(source.model.operatorCodes), 3) + + pipeline = CirclePassManager( + [ + DeadCodeEliminationPass(), + CompactIndicesPass(), + ] + ) + input_value = np.array([1.0, 2.0, 3.0], dtype=np.float32) + expected = (input_value + np.float32(2.0)) * np.float32(3.0) + + result = self.assert_pass_preserves_value( + source, + (input_value,), + lambda document: pipeline.run( + document, + CirclePassContext(verify_after_each_pass=True), + ), + expected_outputs=(expected,), + ) + + transformed = result.document + self.assertEqual(len(transformed.subgraph().operators), 2) + self.assertEqual(len(transformed.subgraph().tensors), 5) + self.assertEqual(len(transformed.model.buffers), 3) + self.assertEqual(len(transformed.model.operatorCodes), 2) + self.assertNotIn( + "dead_output", + [decode_text(tensor.name) for tensor in transformed.subgraph().tensors], + ) + self.assertNotIn( + "dead_addend", + [decode_text(tensor.name) for tensor in transformed.subgraph().tensors], + ) + self.assertTrue(result.transform_result.modified) + self.assertGreater(result.transform_result.changes, 0) + + +if __name__ == "__main__": + import unittest + + unittest.main() diff --git a/test/unit_test/circle/value/test_extract_value.py b/test/unit_test/circle/value/test_extract_value.py new file mode 100644 index 00000000..1d895a34 --- /dev/null +++ b/test/unit_test/circle/value/test_extract_value.py @@ -0,0 +1,108 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import numpy as np + +from tico.circle._schema import decode_text +from tico.circle.operations import extract_by_operator_indices + +from test.support.circle.builder import CircleModelBuilder +from test.support.circle.value_test import CircleValueTestCase + + +class CircleExtractionValueTest(CircleValueTestCase): + """Check extracted graph values against source intermediate tensors.""" + + def test_middle_operator_reproduces_source_boundary_value(self): + """Feed a captured source intermediate into an extracted middle region.""" + + builder = CircleModelBuilder(description="extract-value-test") + x = builder.input("x", [3]) + one = builder.const_f32("one", 1.0) + added = builder.add(x, one, name="added") + two = builder.const_f32("two", 2.0) + multiplied = builder.mul(added, two, name="multiplied") + three = builder.const_f32("three", 3.0) + output = builder.add(multiplied, three, name="output") + builder.set_outputs(output) + source = builder.build() + + input_value = np.array([1.0, 2.0, 3.0], dtype=np.float32) + expected_source = (input_value + np.float32(1.0)) * np.float32( + 2.0 + ) + np.float32(3.0) + result = self.assert_extraction_preserves_value( + source, + (input_value,), + lambda document: extract_by_operator_indices(document, (1,)), + expected_source_outputs=(expected_source,), + ) + + extracted = result.document + self.assertEqual(result.selected_operator_indices, (1,)) + self.assertEqual(result.source_boundary.inputs, (added,)) + self.assertEqual(result.source_boundary.outputs, (multiplied,)) + self.assertEqual(len(extracted.subgraph().operators), 1) + self.assertEqual(len(extracted.subgraph().inputs), 1) + self.assertEqual(len(extracted.subgraph().outputs), 1) + self.assertEqual(len(extracted.model.buffers), 2) + input_tensor = extracted.subgraph().tensors[extracted.subgraph().inputs[0]] + constant_tensor = next( + tensor + for tensor in extracted.subgraph().tensors + if decode_text(tensor.name) == "two" + ) + self.assertEqual(decode_text(input_tensor.name), "added") + self.assertEqual(constant_tensor.buffer, 1) + constant_index = next( + index + for index, tensor in enumerate(extracted.subgraph().tensors) + if tensor is constant_tensor + ) + self.assertNotIn(constant_index, list(extracted.subgraph().inputs)) + + def test_multi_output_region_matches_two_source_intermediates(self): + """Preserve output order for an extracted fan-out region.""" + + builder = CircleModelBuilder(description="extract-multi-output-value-test") + x = builder.input("x", [3]) + one = builder.const_f32("one", 1.0) + shared = builder.add(x, one, name="shared") + two = builder.const_f32("two", 2.0) + doubled = builder.mul(shared, two, name="doubled") + four = builder.const_f32("four", 4.0) + shifted = builder.add(shared, four, name="shifted") + builder.set_outputs(doubled, shifted) + source = builder.build() + + input_value = np.array([1.0, 2.0, 3.0], dtype=np.float32) + result = self.assert_extraction_preserves_value( + source, + (input_value,), + lambda document: extract_by_operator_indices(document, (1, 2)), + expected_source_outputs=( + (input_value + np.float32(1.0)) * np.float32(2.0), + input_value + np.float32(5.0), + ), + ) + + self.assertEqual(result.source_boundary.inputs, (shared,)) + self.assertEqual(result.source_boundary.outputs, (doubled, shifted)) + self.assertEqual(len(result.document.subgraph().outputs), 2) + + +if __name__ == "__main__": + import unittest + + unittest.main() diff --git a/test/unit_test/circle/value/test_io_value.py b/test/unit_test/circle/value/test_io_value.py new file mode 100644 index 00000000..59dde5e2 --- /dev/null +++ b/test/unit_test/circle/value/test_io_value.py @@ -0,0 +1,49 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import numpy as np + +from test.support.circle.builder import CircleModelBuilder +from test.support.circle.value_test import CircleValueTestCase + + +class CircleIOValueTest(CircleValueTestCase): + """Check that Circle serialization preserves executable values.""" + + def test_object_api_round_trip_preserves_output_and_interface(self): + """Compare an Object API fixture with its serialized round trip.""" + + builder = CircleModelBuilder(description="io-value-test") + x = builder.input("x", [2, 3]) + two = builder.const_f32("two", 2.0) + output = builder.add(x, two, name="output") + builder.set_outputs(output) + source = builder.build() + + input_value = np.arange(6, dtype=np.float32).reshape(2, 3) + expected = input_value + np.float32(2.0) + source_result = self.evaluator.evaluate(source, (input_value,)) + restored = self.round_trip(source) + restored_result = self.evaluator.evaluate(restored, (input_value,)) + + self.assert_outputs_equal((expected,), source_result.outputs) + self.assert_outputs_equal((expected,), restored_result.outputs) + self.assert_outputs_equal(source_result.outputs, restored_result.outputs) + self.assert_interfaces_equal(source, restored, check_tensor_names=True) + + +if __name__ == "__main__": + import unittest + + unittest.main() diff --git a/test/unit_test/circle/value/test_layout_ops_value.py b/test/unit_test/circle/value/test_layout_ops_value.py new file mode 100644 index 00000000..302dcadc --- /dev/null +++ b/test/unit_test/circle/value/test_layout_ops_value.py @@ -0,0 +1,106 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import numpy as np + +from tico.circle.passes import CirclePassContext, CirclePassManager +from tico.circle.passes.cleanup import CompactIndicesPass, DeadCodeEliminationPass +from tico.circle.passes.optimization.remove.layout_ops import ( + RemoveRedundantLayoutOpsPass, +) + +from test.support.circle.builder import CircleModelBuilder +from test.support.circle.value_test import CircleValueTestCase + + +class CircleLayoutOpsValueTest(CircleValueTestCase): + """Check numerical equivalence of redundant layout rewrites.""" + + def test_inverse_three_cycle_transposes_are_removed(self): + """Remove inverse transposes whose permutations are not self-inverse.""" + + builder = CircleModelBuilder(description="transpose-value-test") + x = builder.input("x", [2, 3, 4]) + first = builder.transpose(x, [1, 2, 0], name="first_transpose") + output = builder.transpose(first, [2, 0, 1], name="output") + builder.set_outputs(output) + source = builder.build() + + pipeline = CirclePassManager( + [ + RemoveRedundantLayoutOpsPass(), + DeadCodeEliminationPass(), + CompactIndicesPass(), + ] + ) + input_value = np.arange(24, dtype=np.float32).reshape(2, 3, 4) + result = self.assert_pass_preserves_value( + source, + (input_value,), + lambda document: pipeline.run( + document, + CirclePassContext(verify_after_each_pass=True), + ), + expected_outputs=(input_value,), + ) + + transformed = result.document + self.assertEqual(transformed.subgraph().operators, []) + self.assertEqual(transformed.subgraph().inputs, [0]) + self.assertEqual(transformed.subgraph().outputs, [0]) + self.assertEqual(len(transformed.subgraph().tensors), 1) + self.assertEqual(len(transformed.model.buffers), 1) + self.assertEqual(len(transformed.model.operatorCodes), 0) + + def test_consecutive_reshapes_are_reduced_to_the_last_shape(self): + """Bypass the first reshape while preserving the final tensor value.""" + + builder = CircleModelBuilder(description="reshape-value-test") + x = builder.input("x", [2, 3, 4]) + first = builder.reshape(x, [6, 4], name="first_reshape") + output = builder.reshape(first, [4, 6], name="output") + builder.set_outputs(output) + source = builder.build() + + pipeline = CirclePassManager( + [ + RemoveRedundantLayoutOpsPass(), + DeadCodeEliminationPass(), + CompactIndicesPass(), + ] + ) + input_value = np.arange(24, dtype=np.float32).reshape(2, 3, 4) + expected = input_value.reshape(4, 6) + result = self.assert_pass_preserves_value( + source, + (input_value,), + lambda document: pipeline.run( + document, + CirclePassContext(verify_after_each_pass=True), + ), + expected_outputs=(expected,), + ) + + transformed = result.document + self.assertEqual(len(transformed.subgraph().operators), 1) + output_tensor = transformed.subgraph().tensors[ + transformed.subgraph().outputs[0] + ] + self.assertEqual(list(output_tensor.shape), [4, 6]) + + +if __name__ == "__main__": + import unittest + + unittest.main() diff --git a/test/unit_test/circle/value/test_reference_evaluator.py b/test/unit_test/circle/value/test_reference_evaluator.py new file mode 100644 index 00000000..a16c0b84 --- /dev/null +++ b/test/unit_test/circle/value/test_reference_evaluator.py @@ -0,0 +1,206 @@ +# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import unittest + +import numpy as np +from circle_schema import circle + +from test.support.circle.builder import CircleModelBuilder +from test.support.circle.evaluator import CircleReferenceEvaluator + + +class CircleReferenceEvaluatorTest(unittest.TestCase): + """Test the NumPy reference semantics independently of Circle passes.""" + + def setUp(self): + """Create a fresh evaluator for each test.""" + + self.evaluator = CircleReferenceEvaluator() + + def test_add_supports_broadcasting_and_records_intermediates(self): + """Evaluate broadcast ADD and expose its intermediate tensor value.""" + + builder = CircleModelBuilder() + x = builder.input("x", [2, 3]) + bias = builder.const_f32("bias", [1.0, 2.0, 3.0]) + output = builder.add(x, bias, name="output") + builder.set_outputs(output) + document = builder.build() + + input_value = np.arange(6, dtype=np.float32).reshape(2, 3) + expected = input_value + np.array([1.0, 2.0, 3.0], dtype=np.float32) + result = self.evaluator.evaluate(document, (input_value,)) + + np.testing.assert_array_equal(result.outputs[0], expected) + np.testing.assert_array_equal(result.tensor_values[output], expected) + + def test_sub_matches_numpy(self): + """Evaluate SUB with a broadcast constant.""" + + builder = CircleModelBuilder() + x = builder.input("x", [2, 3]) + bias = builder.const_f32("bias", [1.0, 2.0, 3.0]) + output = builder.sub(x, bias, name="output") + builder.set_outputs(output) + document = builder.build() + + input_value = np.arange(6, dtype=np.float32).reshape(2, 3) + expected = input_value - np.array([1.0, 2.0, 3.0], dtype=np.float32) + result = self.evaluator.evaluate(document, (input_value,)) + + np.testing.assert_array_equal(result.outputs[0], expected) + + def test_reshape_and_transpose_match_numpy(self): + """Evaluate a RESHAPE followed by a TRANSPOSE.""" + + builder = CircleModelBuilder() + x = builder.input("x", [2, 3, 4]) + reshaped = builder.reshape(x, [6, 4], name="reshaped") + output = builder.transpose(reshaped, [1, 0], name="output") + builder.set_outputs(output) + document = builder.build() + + input_value = np.arange(24, dtype=np.float32).reshape(2, 3, 4) + expected = input_value.reshape(6, 4).transpose(1, 0) + result = self.evaluator.evaluate(document, (input_value,)) + + np.testing.assert_array_equal(result.outputs[0], expected) + + def test_multiple_outputs_preserve_declared_order(self): + """Return multiple graph outputs in the declared interface order.""" + + builder = CircleModelBuilder() + x = builder.input("x", [3]) + one = builder.const_f32("one", 1.0) + two = builder.const_f32("two", 2.0) + added = builder.add(x, one, name="added") + multiplied = builder.mul(added, two, name="multiplied") + builder.set_outputs(multiplied, added) + document = builder.build() + + input_value = np.array([1.0, 2.0, 3.0], dtype=np.float32) + result = self.evaluator.evaluate(document, (input_value,)) + + np.testing.assert_array_equal(result.outputs[0], (input_value + 1.0) * 2.0) + np.testing.assert_array_equal(result.outputs[1], input_value + 1.0) + + def test_input_shape_mismatch_is_rejected(self): + """Reject inputs that do not match the Circle tensor shape.""" + + builder = CircleModelBuilder() + x = builder.input("x", [2, 3]) + builder.set_outputs(x) + document = builder.build() + + with self.assertRaisesRegex(ValueError, "shape"): + self.evaluator.evaluate( + document, + (np.zeros((3, 2), dtype=np.float32),), + ) + + def test_input_dtype_mismatch_is_rejected(self): + """Reject inputs that do not match the Circle tensor dtype.""" + + builder = CircleModelBuilder() + x = builder.input("x", [2, 3]) + builder.set_outputs(x) + document = builder.build() + + with self.assertRaisesRegex(TypeError, "dtype"): + self.evaluator.evaluate( + document, + (np.zeros((2, 3), dtype=np.float64),), + ) + + def test_unsupported_operator_is_rejected(self): + """Reject builtin operators outside the explicit evaluator subset.""" + + builder = CircleModelBuilder() + x = builder.input("x", [3]) + one = builder.const_f32("one", 1.0) + output = builder.add(x, one, name="output") + builder.set_outputs(output) + document = builder.build() + operator_code = document.model.operatorCodes[0] + operator_code.builtinCode = circle.BuiltinOperator.BuiltinOperator.ABS + operator_code.deprecatedBuiltinCode = circle.BuiltinOperator.BuiltinOperator.ABS + + with self.assertRaisesRegex(NotImplementedError, "builtin operator"): + self.evaluator.evaluate( + document, + (np.ones(3, dtype=np.float32),), + ) + + def test_fused_activation_is_rejected(self): + """Reject arithmetic operators whose fused activation is not NONE.""" + + builder = CircleModelBuilder() + x = builder.input("x", [3]) + one = builder.const_f32("one", 1.0) + output = builder.add(x, one, name="output") + builder.set_outputs(output) + document = builder.build() + document.subgraph().operators[ + 0 + ].builtinOptions.fusedActivationFunction = ( + circle.ActivationFunctionType.ActivationFunctionType.RELU + ) + + with self.assertRaisesRegex(NotImplementedError, "activation"): + self.evaluator.evaluate( + document, + (np.ones(3, dtype=np.float32),), + ) + + def test_optional_input_is_rejected(self): + """Reject an optional operator input instead of guessing semantics.""" + + builder = CircleModelBuilder() + x = builder.input("x", [3]) + one = builder.const_f32("one", 1.0) + output = builder.add(x, one, name="output") + builder.set_outputs(output) + document = builder.build() + document.subgraph().operators[0].inputs[1] = -1 + + with self.assertRaisesRegex(NotImplementedError, "optional inputs"): + self.evaluator.evaluate( + document, + (np.ones(3, dtype=np.float32),), + ) + + def test_external_buffer_is_rejected(self): + """Reject constants backed by an external buffer.""" + + builder = CircleModelBuilder() + x = builder.input("x", [3]) + one = builder.const_f32("one", 1.0) + output = builder.add(x, one, name="output") + builder.set_outputs(output) + document = builder.build() + constant_tensor = document.subgraph().tensors[one] + external_buffer = document.model.buffers[int(constant_tensor.buffer)] + external_buffer.offset = 16 + external_buffer.size = 4 + + with self.assertRaisesRegex(NotImplementedError, "external buffers"): + self.evaluator.evaluate( + document, + (np.ones(3, dtype=np.float32),), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tico/circle/passes/optimization/remove/layout_ops.py b/tico/circle/passes/optimization/remove/layout_ops.py index b923a43c..d81fe865 100644 --- a/tico/circle/passes/optimization/remove/layout_ops.py +++ b/tico/circle/passes/optimization/remove/layout_ops.py @@ -16,6 +16,7 @@ from typing import Any, TYPE_CHECKING +from tico.circle._schema import circle_schema from tico.circle.graph import as_indices, as_list from tico.circle.passes.base import CirclePass, CirclePassContext, CirclePassResult from tico.circle.rewrite import replace_tensor_uses @@ -24,60 +25,67 @@ from tico.circle.document import CircleDocument -def _get_builtin_code(operator: Any) -> int: - """Extract the builtin operator code from an operator.""" - builtin_options = getattr(operator, "builtinOptions", None) - if builtin_options is None: - return -1 - # The builtin operator type is stored in the BuiltinOperator enum - # We need to extract it from the operator's BuiltinOptions - builtin_ops = getattr(operator, "opcodeIndex", -1) - return int(builtin_ops) if builtin_ops is not None else -1 +def _builtin_operator_value(name: str) -> int: + """Return a builtin operator value from the generated Circle schema.""" + + schema = circle_schema() + enum_module = getattr(schema, "BuiltinOperator", None) + enum_type = ( + getattr(enum_module, "BuiltinOperator", None) + if enum_module is not None + else None + ) + if enum_type is None or not hasattr(enum_type, name): + raise RuntimeError(f"Circle schema does not provide BuiltinOperator.{name}.") + return int(getattr(enum_type, name)) + + +_RESHAPE_BUILTIN_CODE = _builtin_operator_value("RESHAPE") +_TRANSPOSE_BUILTIN_CODE = _builtin_operator_value("TRANSPOSE") def _is_reshape_op(operator: Any, operator_codes: list[Any]) -> bool: - """Check if operator is a Reshape operation.""" + """Return whether an operator references the Circle RESHAPE builtin.""" + try: opcode_index = int(getattr(operator, "opcodeIndex", -1)) if opcode_index < 0 or opcode_index >= len(operator_codes): return False opcode = operator_codes[opcode_index] builtin_code = int(getattr(opcode, "builtinCode", -1)) - # BuiltinOperator::RESHAPE = 26 - return builtin_code == 26 + return builtin_code == _RESHAPE_BUILTIN_CODE except (TypeError, ValueError, AttributeError): return False def _is_transpose_op(operator: Any, operator_codes: list[Any]) -> bool: - """Check if operator is a Transpose operation.""" + """Return whether an operator references the Circle TRANSPOSE builtin.""" + try: opcode_index = int(getattr(operator, "opcodeIndex", -1)) if opcode_index < 0 or opcode_index >= len(operator_codes): return False opcode = operator_codes[opcode_index] builtin_code = int(getattr(opcode, "builtinCode", -1)) - # BuiltinOperator::TRANSPOSE = 54 - return builtin_code == 54 + return builtin_code == _TRANSPOSE_BUILTIN_CODE except (TypeError, ValueError, AttributeError): return False def _check_perm(first_perm: list[int], second_perm: list[int]) -> bool: - """Check if first_perm[second_perm[i]] == i for all i. + """Return whether composing two permutations produces the identity.""" - This verifies if composing two permutations results in identity. - """ if len(first_perm) != len(second_perm): return False - for i in range(len(second_perm)): - if first_perm[second_perm[i]] != i: + for index in range(len(second_perm)): + if first_perm[second_perm[index]] != index: return False return True def _get_const_data(graph: Any, tensor_index: int) -> list[int] | None: - """Extract constant data from a tensor, returns list of ints or None.""" + """Decode an inline INT32 constant tensor as a list of Python integers.""" + tensors = as_list(graph.subgraph.tensors) if tensor_index < 0 or tensor_index >= len(tensors): return None @@ -100,13 +108,12 @@ def _get_const_data(graph: Any, tensor_index: int) -> list[int] | None: return None try: - # Convert buffer data to list of ints import struct result = [] - for i in range(0, len(data), 4): - if i + 4 <= len(data): - value = struct.unpack(" list[int] | None: class RemoveRedundantLayoutOpsPass(CirclePass): - """Remove redundant Reshape and Transpose operations. + """Remove redundant consecutive Reshape and Transpose operations. - This pass optimizes consecutive Reshape or Transpose operations that - either cancel each other out or can be fused into a single operation. + The pass handles two patterns: - Handles two patterns: - 1. Reshape-Reshape: Multiple reshapes can be fused or simplified - 2. Transpose-Transpose: Consecutive transposes can be fused or eliminated + 1. Consecutive Reshape operators, where the second Reshape can consume the + first Reshape input directly. + 2. Consecutive inverse Transpose operators, where the pair can be bypassed. """ def run( @@ -129,7 +135,7 @@ def run( document: CircleDocument, context: CirclePassContext, ) -> CirclePassResult: - """Remove redundant layout operations from all subgraphs.""" + """Remove redundant layout operations from every subgraph.""" changes = 0 @@ -142,17 +148,17 @@ def run( if not operators: continue - # Process operators: reshape operations for operator in operators: if _is_reshape_op(operator, operator_codes): if self._remove_redundant_reshape(graph, operator): changes += 1 - # Process operators: transpose operations for operator in operators: if _is_transpose_op(operator, operator_codes): if self._remove_redundant_transpose( - graph, operator, operator_codes + graph, + operator, + operator_codes, ): changes += 1 @@ -166,72 +172,50 @@ def run( ) def _remove_redundant_reshape(self, graph: Any, reshape_op: Any) -> bool: - """Remove redundant consecutive Reshape operations. + """Bypass the first operator in a consecutive Reshape pair. - Pattern: - input → Reshape → Reshape → output + Pattern:: - Simplification: - input → Reshape → output + input -> Reshape -> Reshape -> output - Args: - graph: CircleGraph object with model structure - reshape_op: Current Reshape operator + Simplification:: - Returns: - True if optimization was applied, False otherwise + input -----------> Reshape -> output """ + inputs = as_indices(getattr(reshape_op, "inputs", None)) - if len(inputs) < 1: + if not inputs: return False input_tensor = inputs[0] producer = graph.producer(input_tensor) - if producer is None: return False operators = as_list(graph.subgraph.operators) operator_codes = as_list(graph.model.operatorCodes) - if producer >= len(operators): return False producer_op = operators[producer] - - # Check if producer is also a Reshape if not _is_reshape_op(producer_op, operator_codes): return False - # Get producer's input producer_inputs = as_indices(getattr(producer_op, "inputs", None)) - if len(producer_inputs) < 1: + if not producer_inputs: return False - # Connect current reshape to producer's input, skipping intermediate reshape reshape_op.inputs = [producer_inputs[0]] + list(inputs[1:]) return True def _remove_redundant_transpose( - self, graph: Any, transpose_op: Any, operator_codes: list[Any] + self, + graph: Any, + transpose_op: Any, + operator_codes: list[Any], ) -> bool: - """Remove or fuse consecutive Transpose operations. + """Bypass two consecutive inverse Transpose operators.""" - Patterns: - 1. Inverse transposes cancel out: - Transpose(perm1) → Transpose(perm2) where perm1[perm2[i]] == i - - 2. General composition: - Transpose(perm1) → Transpose(perm2) → Transpose(composite) - - Args: - graph: CircleGraph object with model structure - transpose_op: Current Transpose operator - operator_codes: List of operator codes from model - - Returns: - True if optimization was applied, False otherwise - """ inputs = as_indices(getattr(transpose_op, "inputs", None)) if len(inputs) < 2: return False @@ -241,21 +225,16 @@ def _remove_redundant_transpose( return False input_tensor = inputs[0] - perm_tensor = inputs[1] - + permutation_tensor = inputs[1] producer = graph.producer(input_tensor) - if producer is None: return False operators = as_list(graph.subgraph.operators) - if producer >= len(operators): return False producer_op = operators[producer] - - # Check if producer is also a Transpose if not _is_transpose_op(producer_op, operator_codes): return False @@ -263,20 +242,16 @@ def _remove_redundant_transpose( if len(producer_inputs) < 2: return False - producer_perm_tensor = producer_inputs[1] - - # Extract permutation data - perm_data = _get_const_data(graph, perm_tensor) - producer_perm_data = _get_const_data(graph, producer_perm_tensor) - - if perm_data is None or producer_perm_data is None: + producer_permutation_tensor = producer_inputs[1] + permutation = _get_const_data(graph, permutation_tensor) + producer_permutation = _get_const_data( + graph, + producer_permutation_tensor, + ) + if permutation is None or producer_permutation is None: return False - # Check if the composition is identity (inverse transpose) - if _check_perm(perm_data, producer_perm_data): - # Bypass the inverse pair at the second Transpose output. Dead-code - # elimination removes the second Transpose and removes the first one - # when it has no other consumers. + if _check_perm(permutation, producer_permutation): replacement = replace_tensor_uses( graph.model, subgraph_index=graph.subgraph_index, @@ -285,6 +260,4 @@ def _remove_redundant_transpose( ) return replacement.modified - # TODO: Implement general composition optimization - # This would create a new permutation constant for the composite transpose return False