Add Transformer API Interface and @cirq.transformer decorator

cirq-core/cirq/__init__.py

@@ -365,7 +365,11 @@
     single_qubit_matrix_to_phased_x_z,
     single_qubit_matrix_to_phxz,
     single_qubit_op_to_framed_phase_form,
+    TRANSFORMER,
+    TransformerContext,
+    TransformerLogger,
     three_qubit_matrix_to_operations,
+    transformer,
     two_qubit_matrix_to_diagonal_and_operations,
     two_qubit_matrix_to_operations,
     two_qubit_matrix_to_sqrt_iswap_operations,

cirq-core/cirq/protocols/json_test_data/spec.py

@@ -110,6 +110,9 @@
         'MergeSingleQubitGates',
         'PointOptimizer',
         'SynchronizeTerminalMeasurements',
+        # Transformers
+        'TransformerLogger',
+        'TransformerContext',
         # global objects
         'CONTROL_TAG',
         'PAULI_BASIS',
@@ -172,6 +175,7 @@
         'Sweepable',
         'TParamKey',
         'TParamVal',
+        'TRANSFORMER',
         'ParamDictType',
         # utility:
         'CliffordSimulator',

cirq-core/cirq/transformers/__init__.py

@@ -41,6 +41,14 @@
     two_qubit_gate_product_tabulation,
 )
 
+from cirq.transformers.transformer_api import (
+    LogLevel,
+    TRANSFORMER,
+    TransformerContext,
+    TransformerLogger,
+    transformer,
+)
+
 from cirq.transformers.transformer_primitives import (
     map_moments,
     map_operations,

cirq-core/cirq/transformers/transformer_api.py

@@ -0,0 +1,313 @@
+# Copyright 2022 The Cirq Developers
+#
+# 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
+#
+#     https://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.
+
+"""Defines the API for circuit transformers in Cirq."""
+
+import textwrap
+import functools
+from typing import (
+    Any,
+    Callable,
+    Tuple,
+    Hashable,
+    List,
+    Type,
+    overload,
+    TYPE_CHECKING,
+)
+import dataclasses
+import enum
+from cirq.circuits.circuit import CIRCUIT_TYPE
+
+if TYPE_CHECKING:
+    import cirq
+
+
+class LogLevel(enum.Enum):
+    """Different logging resolution options for `cirq.TransformerLogger`.
+
+    The enum values of the logging levels are used to filter the stored logs when printing.
+    In general, a logging level `X` includes all logs stored at a level >= 'X'.
+
+    Args:
+        ALL:     All levels. Used to filter logs when printing.
+        DEBUG:   Designates fine-grained informational events that are most useful to debug /
+                 understand in-depth any unexpected behavior of the transformer.
+        INFO:    Designates informational messages that highlight the actions of a transformer.
+        WARNING: Designates unwanted or potentially harmful situations.
+        NONE:    No levels. Used to filter logs when printing.
+    """
+
+    ALL = 0
+    DEBUG = 10
+    INFO = 20
+    WARNING = 30
+    NONE = 40
+
+
+@dataclasses.dataclass
+class _LoggerNode:
+    """Stores logging data of a single transformer stage in `cirq.TransformerLogger`.
+
+    The class is used to define a logging graph to store logs of sequential or nested transformers.
+    Each node corresponds to logs of a single transformer stage.
+
+    Args:
+        transformer_id:   Integer specifying a unique id for corresponding transformer stage.
+        transformer_name: Name of the corresponding transformer stage.
+        initial_circuit:  Initial circuit before the transformer stage began.
+        final_circuit:    Final circuit after the transformer stage ended.
+        logs:             Messages logged by the transformer stage.
+        nested_loggers:   `transformer_id`s of nested transformer stages which were called by
+                          the current stage.
+    """
+
+    transformer_id: int
+    transformer_name: str
+    initial_circuit: 'cirq.AbstractCircuit'
+    final_circuit: 'cirq.AbstractCircuit'
+    logs: List[Tuple[LogLevel, Tuple[str, ...]]] = dataclasses.field(default_factory=list)
+    nested_loggers: List[int] = dataclasses.field(default_factory=list)
+
+
+class TransformerLogger:
+    """Base Class for transformer logging infrastructure. Defaults to text-based logging.
+
+    The logger implementation should be stateful, s.t.:
+        - Each call to `register_initial` registers a new transformer stage and initial circuit.
+        - Each subsequent call to `log` should store additional logs corresponding to the stage.
+        - Each call to `register_final` should register the end of the currently active stage.
+
+    The logger assumes that
+        - Transformers are run sequentially.
+        - Nested transformers are allowed, in which case the behavior would be similar to a
+          doing a depth first search on the graph of transformers -- i.e. the top level transformer
+          would end (i.e. receive a `register_final` call) once all nested transformers (i.e. all
+          `register_initial` calls received while the top level transformer was active) have
+          finished (i.e. corresponding `register_final` calls have also been received).
+        - This behavior can be simulated by maintaining a stack of currently active stages and
+          adding data from `log` calls to the stage at the top of the stack.
+
+    The `LogLevel`s can be used to control the input processing and output resolution of the logs.
+    """
+
+    def __init__(self):
+        """Initializes TransformerLogger."""
+        self._curr_id: int = 0
+        self._logs: List[_LoggerNode] = []
+        self._stack: List[int] = []
+
+    def register_initial(self, circuit: 'cirq.AbstractCircuit', transformer_name: str) -> None:
+        """Register the beginning of a new transformer stage.
+
+        Args:
+            circuit: Input circuit to the new transformer stage.
+            transformer_name: Name of the new transformer stage.
+        """
+        if self._stack:
+            self._logs[self._stack[-1]].nested_loggers.append(self._curr_id)
+        self._logs.append(_LoggerNode(self._curr_id, transformer_name, circuit, circuit))
+        self._stack.append(self._curr_id)
+        self._curr_id += 1
+
+    def log(self, *args: str, level: LogLevel = LogLevel.INFO) -> None:
+        """Log additional metadata corresponding to the currently active transformer stage.
+
+        Args:
+            *args: The additional metadata to log.
+            level: Logging level to control the amount of metadata that gets put into the context.
+
+        Raises:
+            ValueError: If there's no active transformer on the stack.
+        """
+        if len(self._stack) == 0:
+            raise ValueError('No active transformer found.')
+        self._logs[self._stack[-1]].logs.append((level, args))
+
+    def register_final(self, circuit: 'cirq.AbstractCircuit', transformer_name: str) -> None:
+        """Register the end of the currently active transformer stage.
+
+        Args:
+            circuit: Final transformed output circuit from the transformer stage.
+            transformer_name: Name of the (currently active) transformer stage which ends.
+
+        Raises:
+            ValueError: If `transformer_name` is different from currently active transformer name.
+        """
+        tid = self._stack.pop()
+        if self._logs[tid].transformer_name != transformer_name:
+            raise ValueError(
+                f"Expected `register_final` call for currently active transformer "
+                f"{self._logs[tid].transformer_name}."
+            )
+        self._logs[tid].final_circuit = circuit
+
+    def show(self, level: LogLevel = LogLevel.INFO) -> None:
+        """Show the stored logs >= level in the desired format.
+
+        Args:
+            level: The logging level to filter the logs with. The method shows all logs with a
+            `LogLevel` >= `level`.
+        """
+
+        def print_log(log: _LoggerNode, pad=''):
+            print(pad, f"Transformer-{1+log.transformer_id}: {log.transformer_name}", sep='')
+            print(pad, "Initial Circuit:", sep='')
+            print(textwrap.indent(str(log.initial_circuit), pad), "\n", sep='')
+            for log_level, log_text in log.logs:
+                if log_level.value >= level.value:
+                    print(pad, log_level, *log_text)
+            print("\n", pad, "Final Circuit:", sep='')
+            print(textwrap.indent(str(log.final_circuit), pad))
+            print("----------------------------------------")
+
+        done = [0] * self._curr_id
+        for i in range(self._curr_id):
+            # Iterative DFS.
+            stack = [(i, '')] if not done[i] else []
+            while len(stack) > 0:
+                log_id, pad = stack.pop()
+                print_log(self._logs[log_id], pad)
+                done[log_id] = True
+                for child_id in self._logs[log_id].nested_loggers[::-1]:
+                    stack.append((child_id, pad + ' ' * 4))
+
+
+class NoOpTransformerLogger(TransformerLogger):
+    """All calls to this logger are a no-op"""
+
+    def register_initial(self, circuit: 'cirq.AbstractCircuit', transformer_name: str) -> None:
+        pass
+
+    def log(self, *args: str, level: LogLevel = LogLevel.INFO) -> None:
+        pass
+
+    def register_final(self, circuit: 'cirq.AbstractCircuit', transformer_name: str) -> None:
+        pass
+
+    def show(self, level: LogLevel = LogLevel.INFO) -> None:
+        pass
+
+
+@dataclasses.dataclass()
+class TransformerContext:
+    """Stores common configurable options for transformers.
+
+    Args:
+        logger: `cirq.TransformerLogger` instance, which is a stateful logger used for logging
+                the actions of individual transformer stages. The same logger instance should be
+                shared across different transformer calls.
+        ignore_tags: Tuple of tags which should be ignored while applying transformations on a
+                circuit. Transformers should not transform any operation marked with a tag that
+                belongs to this tuple. Note that any instance of a Hashable type (like `str`,
+                `cirq.VirtualTag` etc.) is a valid tag.
+    """
+
+    logger: TransformerLogger = NoOpTransformerLogger()
+    ignore_tags: Tuple[Hashable, ...] = ()
+
+
+TRANSFORMER = Callable[['cirq.AbstractCircuit', TransformerContext], 'cirq.AbstractCircuit']
+_TRANSFORMER_TYPE = Callable[['cirq.AbstractCircuit', TransformerContext], CIRCUIT_TYPE]
+
+
+def _transform_and_log(
+    func: _TRANSFORMER_TYPE[CIRCUIT_TYPE],
+    transformer_name: str,
+    circuit: 'cirq.AbstractCircuit',
+    context: TransformerContext,
+) -> CIRCUIT_TYPE:
+    """Helper to log initial and final circuits before and after calling the transformer."""
+
+    context.logger.register_initial(circuit, transformer_name)
+    transformed_circuit = func(circuit, context)
+    context.logger.register_final(transformed_circuit, transformer_name)
+    return transformed_circuit
+
+
+def _transformer_class(
+    cls: Type[_TRANSFORMER_TYPE[CIRCUIT_TYPE]],
+) -> Type[_TRANSFORMER_TYPE[CIRCUIT_TYPE]]:
+    old_func = cls.__call__
+
+    def transformer_with_logging_cls(
+        self: Type[_TRANSFORMER_TYPE[CIRCUIT_TYPE]],
+        circuit: 'cirq.AbstractCircuit',
+        context: TransformerContext,
+    ) -> CIRCUIT_TYPE:
+        def call_old_func(c: 'cirq.AbstractCircuit', ct: TransformerContext) -> CIRCUIT_TYPE:
+            return old_func(self, c, ct)
+
+        return _transform_and_log(call_old_func, cls.__name__, circuit, context)
+
+    setattr(cls, '__call__', transformer_with_logging_cls)
+    return cls
+
+
+def _transformer_func(func: _TRANSFORMER_TYPE[CIRCUIT_TYPE]) -> _TRANSFORMER_TYPE[CIRCUIT_TYPE]:
+    @functools.wraps(func)
+    def transformer_with_logging_func(
+        circuit: 'cirq.AbstractCircuit',
+        context: TransformerContext,
+    ) -> CIRCUIT_TYPE:
+        return _transform_and_log(func, func.__name__, circuit, context)
+
+    return transformer_with_logging_func
+
+
+@overload
+def transformer(cls_or_func: _TRANSFORMER_TYPE[CIRCUIT_TYPE]) -> _TRANSFORMER_TYPE[CIRCUIT_TYPE]:
+    pass
+
+
+@overload
+def transformer(
+    cls_or_func: Type[_TRANSFORMER_TYPE[CIRCUIT_TYPE]],
+) -> Type[_TRANSFORMER_TYPE[CIRCUIT_TYPE]]:
+    pass
+
+
+def transformer(cls_or_func: Any) -> Any:
+    """Decorator to verify API and append logging functionality to transformer functions & classes.
+
+    The decorated function or class must satisfy
+    `Callable[[cirq.Circuit, cirq.TransformerContext], cirq.Circuit]` API. For Example:
+
+    >>> @cirq.transformer
+    >>> def convert_to_cz(circuit: cirq.Circuit, context: cirq.TransformerContext) -> cirq.Circuit:
+    >>>    ...
+
+    The decorated class must implement the `__call__` method to satisfy the above API.
+
+    >>> @cirq.transformer
+    >>> class ConvertToSqrtISwaps:
+    >>>    def __init__(self):
+    >>>        ...
+    >>>    def __call__(
+    >>>        self, circuit: cirq.Circuit, context: cirq.TransformerContext
+    >>>    ) -> cirq.Circuit:
+    >>>        ...
+
+    Args:
+        cls_or_func: The callable class or method to be decorated.
+
+    Returns:
+        Decorated class / method which includes additional logging boilerplate. The decorated
+        callable always receives a copy of the input circuit so that the input is never mutated.
+    """
+    if isinstance(cls_or_func, type):
+        return _transformer_class(cls_or_func)
+    else:
+        assert callable(cls_or_func)
+        return _transformer_func(cls_or_func)