CQT-250 add barrier directives

opensquirrel/circuit_builder.py

@@ -9,24 +9,16 @@
 from typing_extensions import Self
 
 from opensquirrel.circuit import Circuit
+from opensquirrel.default_directives import default_directive_set
 from opensquirrel.default_gates import default_gate_aliases, default_gate_set
 from opensquirrel.default_measures import default_measure_set
 from opensquirrel.default_resets import default_reset_set
-from opensquirrel.instruction_library import GateLibrary, MeasureLibrary, ResetLibrary
-from opensquirrel.ir import (
-    ANNOTATIONS_TO_TYPE_MAP,
-    IR,
-    Comment,
-    Gate,
-    Measure,
-    Qubit,
-    QubitLike,
-    Reset,
-)
+from opensquirrel.instruction_library import DirectiveLibrary, GateLibrary, MeasureLibrary, ResetLibrary
+from opensquirrel.ir import ANNOTATIONS_TO_TYPE_MAP, IR, Comment, Directive, Gate, Measure, Qubit, QubitLike, Reset
 from opensquirrel.register_manager import BitRegister, QubitRegister, RegisterManager
 
 
-class CircuitBuilder(GateLibrary, MeasureLibrary, ResetLibrary):
+class CircuitBuilder(GateLibrary, MeasureLibrary, ResetLibrary, DirectiveLibrary):
     """
     A class using the builder pattern to make construction of circuits easy from Python.
     Adds corresponding instruction when a method is called. Checks that instructions are known and called with the right
@@ -44,7 +36,7 @@ class CircuitBuilder(GateLibrary, MeasureLibrary, ResetLibrary):
         >>> CircuitBuilder(qubit_register_size=3, bit_register_size=3).\
         H(0).CNOT(0, 1).CNOT(0, 2).\
         to_circuit()
-        version 3.0
+        version 3.
         <BLANKLINE>
         qubit[3] q
         <BLANKLINE>
@@ -63,10 +55,12 @@ def __init__(
         gate_aliases: Mapping[str, Callable[..., Gate]] = default_gate_aliases,
         measure_set: list[Callable[..., Measure]] = default_measure_set,
         reset_set: list[Callable[..., Reset]] = default_reset_set,
+        directive_set: list[Callable[..., Directive]] = default_directive_set,
     ) -> None:
         GateLibrary.__init__(self, gate_set, gate_aliases)
         MeasureLibrary.__init__(self, measure_set)
         ResetLibrary.__init__(self, reset_set)
+        DirectiveLibrary.__init__(self, directive_set)
         self.register_manager = RegisterManager(QubitRegister(qubit_register_size), BitRegister(bit_register_size))
         self.ir = IR()
 
@@ -89,6 +83,10 @@ def _add_instruction(self, attr: str, *args: Any) -> Self:
             generator_f_reset = ResetLibrary.get_reset_f(self, attr)
             self._check_generator_f_args(generator_f_reset, attr, args)
             self.ir.add_reset(generator_f_reset(*args))
+        elif any(attr == directive.__name__ for directive in self.directive_set):
+            generator_f_directive = DirectiveLibrary.get_directive_f(self, attr)
+            self._check_generator_f_args(generator_f_directive, attr, args)
+            self.ir.add_directive(generator_f_directive(*args))
         else:
             generator_f_gate = GateLibrary.get_gate_f(self, attr)
             self._check_generator_f_args(generator_f_gate, attr, args)
@@ -118,7 +116,7 @@ def _check_bit_out_of_bounds_access(self, index: int) -> None:
 
     def _check_generator_f_args(
         self,
-        generator_f: Callable[..., Gate | Measure | Reset],
+        generator_f: Callable[..., Gate | Measure | Reset | Directive],
         attr: str,
         args: tuple[Any, ...],
     ) -> None:

opensquirrel/default_directives.py

@@ -0,0 +1,9 @@
+from opensquirrel.ir import Barrier, QubitLike, named_directive
+
+
+@named_directive
+def barrier(q: QubitLike) -> Barrier:
+    return Barrier(q)
+
+
+default_directive_set = [barrier]

opensquirrel/default_gates.py

@@ -152,13 +152,7 @@ def CRk(  # noqa: N802
     Rz,
 ]
 default_gate_set: list[Callable[..., Gate]]
-default_gate_set = [
-    *default_bloch_sphere_rotations,
-    CNOT,
-    CZ,
-    CR,
-    CRk,
-]
+default_gate_set = [*default_bloch_sphere_rotations, CNOT, CZ, CR, CRk]
 
 default_gate_aliases = {
     "Hadamard": H,

opensquirrel/instruction_library.py

@@ -4,7 +4,7 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from opensquirrel.ir import Gate, Measure, Reset
+    from opensquirrel.ir import Directive, Gate, Measure, Reset
 
 
 class InstructionLibrary:
@@ -53,3 +53,15 @@ def get_reset_f(self, reset_name: str) -> Callable[..., Reset]:
         except StopIteration as exc:
             msg = f"unknown instruction `{reset_name}`"
             raise ValueError(msg) from exc
+
+
+class DirectiveLibrary(InstructionLibrary):
+    def __init__(self, directive_set: Iterable[Callable[..., Directive]]) -> None:
+        self.directive_set = directive_set
+
+    def get_directive_f(self, directive_name: str) -> Callable[..., Directive]:
+        try:
+            return next(f for f in self.directive_set if f.__name__ == directive_name)
+        except StopIteration as exc:
+            msg = f"unknown instruction `{directive_name}`"
+            raise ValueError(msg) from exc

opensquirrel/ir.py

@@ -63,6 +63,12 @@ def visit_matrix_gate(self, matrix_gate: MatrixGate) -> Any:
     def visit_controlled_gate(self, controlled_gate: ControlledGate) -> Any:
         pass
 
+    def visit_directive(self, directive: Directive) -> Any:
+        pass
+
+    def visit_barrier(self, barrier: Barrier) -> Any:
+        pass
+
 
 class IRNode(ABC):
     @abstractmethod
@@ -365,6 +371,66 @@ def get_qubit_operands(self) -> list[Qubit]:
         return [self.qubit]
 
 
+class Directive(Statement, ABC):
+    def __init__(
+        self,
+        generator: Callable[..., Directive] | None = None,
+        arguments: tuple[Expression, ...] | None = None,
+    ) -> None:
+        self.generator = generator
+        self.arguments = arguments
+
+    @property
+    def name(self) -> str:
+        if self.generator:
+            return self.generator.__name__
+        return "Anonymous directive: " + self.__repr__()
+
+    @property
+    def is_abstract(self) -> bool:
+        return self.arguments is None
+
+    @abstractmethod
+    def get_qubit_operands(self) -> list[Qubit]:
+        """Get the qubit operands of the Gate.
+
+        Returns:
+            List of qubits on which the Gate operates.
+        """
+
+
+class Barrier(Directive):
+    def __init__(
+        self,
+        qubit: QubitLike,
+        generator: Callable[..., Barrier] | None = None,
+        arguments: tuple[Expression, ...] | None = None,
+    ) -> None:
+        Directive.__init__(self, generator, arguments)
+        self.generator = generator
+        self.arguments = arguments
+        self.qubit = Qubit(qubit)
+
+    def __repr__(self) -> str:
+        return f"Barrier(qubit={self.qubit})"
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, Barrier):
+            return False
+
+        return self.qubit == other.qubit
+
+    def accept(self, visitor: IRVisitor) -> Any:
+        visitor.visit_directive(self)
+        return visitor.visit_barrier(self)
+
+    def get_qubit_operands(self) -> list[Qubit]:
+        return [self.qubit]
+
+    def is_identity(self) -> bool:
+        return False
+
+
 class Gate(Statement, ABC):
     def __init__(
         self,
@@ -643,6 +709,32 @@ def wrapper(*args: Any, **kwargs: Any) -> Reset:
     return wrapper
 
 
+def named_directive(directive_generator: Callable[..., Directive]) -> Callable[..., Directive]:
+    @wraps(directive_generator)
+    def wrapper(*args: Any, **kwargs: Any) -> Directive:
+        result = directive_generator(*args, **kwargs)
+        result.generator = wrapper
+
+        all_args: list[Any] = []
+        for par in inspect.signature(directive_generator).parameters.values():
+            next_arg = kwargs[par.name] if par.name in kwargs else args[len(all_args)]
+            next_annotation = (
+                ANNOTATIONS_TO_TYPE_MAP[par.annotation] if isinstance(par.annotation, str) else par.annotation
+            )
+
+            # Convert to correct expression for IR
+            if is_qubit_like_annotation(next_annotation):
+                next_arg = Qubit(next_arg)
+
+            # Append parsed argument
+            all_args.append(next_arg)
+
+        result.arguments = tuple(all_args)
+        return result
+
+    return wrapper
+
+
 def compare_gates(g1: Gate, g2: Gate) -> bool:
     union_mapping = [q.index for q in list(set(g1.get_qubit_operands()) | set(g2.get_qubit_operands()))]
 
@@ -685,6 +777,9 @@ def add_reset(self, reset: Reset) -> None:
     def add_comment(self, comment: Comment) -> None:
         self.statements.append(comment)
 
+    def add_directive(self, directive: Directive) -> None:
+        self.statements.append(directive)
+
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, IR):
             return False

opensquirrel/merger/general_merger.py

@@ -1,11 +1,18 @@
+from __future__ import annotations
+
+import functools
+import operator
 from math import acos, cos, floor, log10, sin
+from typing import TYPE_CHECKING, Any
 
 import numpy as np
 
-from opensquirrel.circuit import Circuit
 from opensquirrel.common import ATOL
 from opensquirrel.default_gates import I, default_bloch_sphere_rotations_without_params
-from opensquirrel.ir import BlochSphereRotation, Comment, Qubit
+from opensquirrel.ir import Barrier, BlochSphereRotation, Comment, Qubit, Statement
+
+if TYPE_CHECKING:
+    from opensquirrel.circuit import Circuit
 
 
 def compose_bloch_sphere_rotations(a: BlochSphereRotation, b: BlochSphereRotation) -> BlochSphereRotation:
@@ -79,6 +86,41 @@ def try_name_anonymous_bloch(bsr: BlochSphereRotation) -> BlochSphereRotation:
     return bsr
 
 
+def _flatten_list(list_to_flatten: list[list[Any]]) -> list[Any]:
+    return functools.reduce(operator.iadd, list_to_flatten, [])
+
+
+def merge_barriers(statement_list: list[Statement]) -> list[Statement]:
+    """Receives a list of statements.
+    Returns an ordered version of the input list of statements where groups of barriers are merged together,
+    and placed as late in the list as possible.
+
+    Args:
+        statement_list: list of statements
+
+    Returns:
+        Statement list with the barriers merged.
+    """
+    barrier_list: list[Barrier] = []
+    ordered_statement_list: list[Statement] = []
+    for _, statement in enumerate(statement_list):
+        if isinstance(statement, Barrier):
+            barrier_list.append(statement)
+        else:
+            if len(barrier_list) > 0 and hasattr(statement, "get_qubit_operands"):
+                instruction_qubits = statement.get_qubit_operands()
+                barrier_qubits = _flatten_list([barrier.get_qubit_operands() for barrier in barrier_list])
+                if any(qubit in barrier_qubits for qubit in instruction_qubits):
+                    ordered_statement_list.extend(barrier_list)
+                    barrier_list = []
+            ordered_statement_list.append(statement)
+
+    if len(barrier_list) > 0:
+        ordered_statement_list.extend(barrier_list)
+
+    return ordered_statement_list
+
+
 def merge_single_qubit_gates(circuit: Circuit) -> None:
     """Merge all consecutive 1-qubit gates in the circuit.
 
@@ -126,3 +168,5 @@ def merge_single_qubit_gates(circuit: Circuit) -> None:
             if accumulated_bloch_sphere_rotation.is_anonymous:
                 accumulated_bloch_sphere_rotation = try_name_anonymous_bloch(accumulated_bloch_sphere_rotation)
             ir.statements.append(accumulated_bloch_sphere_rotation)
+
+    ir.statements = merge_barriers(circuit.ir.statements)

opensquirrel/writer/writer.py

@@ -5,6 +5,7 @@
 from opensquirrel.ir import (
     Bit,
     Comment,
+    Directive,
     Float,
     Gate,
     Int,
@@ -65,6 +66,13 @@ def visit_reset(self, reset: Reset) -> None:
         qubit_argument = reset.arguments[0].accept(self)  # type: ignore[index]
         self.output += f"{reset.name} {qubit_argument}\n"
 
+    def visit_directive(self, directive: Directive) -> None:
+        if directive.is_abstract:
+            self.output += f"{directive.name}\n"
+            return
+        qubit_argument = directive.arguments[0].accept(self)  # type: ignore[index]
+        self.output += f"{directive.name} {qubit_argument}\n"
+
     def visit_gate(self, gate: Gate) -> None:
         gate_name = gate.name
         gate_generator = []