Add TICK command and edge scheduler for temporal scheduling

CHANGELOG.md

@@ -7,6 +7,34 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- **TICK Command**: Time slice boundary marker for temporal scheduling in MBQC patterns
+  - Added TICK command type to mark boundaries between time slices
+  - Integrated TICK command handling in PatternSimulator
+  - Integrated TICK command processing in Stim compiler
+
+- **Edge Scheduler**: Automatic entanglement operation scheduling based on node preparation times ([#99](https://github.com/TeamGraphix/graphqomb/issues/99))
+  - Added `entangle_time` attribute to Scheduler for tracking entanglement operation timing
+  - Added `auto_schedule_entanglement()` method to automatically schedule CZ gates when both nodes are prepared
+  - Extended the `timeline` property to include entanglement operations
+  - Added entanglement time validation in schedule validation
+  - Added `compress_schedule()` function to support entanglement time compression
+
+- **Pattern**: Added the `depth` attribute into `Pattern`, which represents the depth of parallel execution.
+
+- **Scheduler Integration**: Enhanced qompile() to support temporal scheduling with TICK commands
+  - Added `scheduler` parameter to qompile() for custom scheduling
+  - Automatically inserts TICK commands between time slices
+
+- **Examples**: Added entanglement_scheduling_demo.py demonstrating edge scheduler features
+
+### Changed
+
+- **Pattern**: Updated command sequence generation to support TICK commands
+- **Command**: Extended Command type alias to include TICK
+- The default strategy of `Scheduler.solve_schedule` is now `MINIMIZE_TIME` instead of `MINIMIZE_SPACE` for the compilation performance.
+
 ## [0.1.2] - 2025-10-31
 
 ### Added
@@ -25,6 +53,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Tests
 
+- **TICK Command**: Added comprehensive test suite for TICK command functionality
+  - Added `test_simulator_with_tick_commands()` for TICK command handling in PatternSimulator
+  - Added `test_stim_compile_with_tick_commands()` for TICK command compilation to Stim format
+  - Extended scheduler integration tests with comprehensive edge scheduling validation
+
 - **Pauli Frame**: Added comprehensive test suite for PauliFrame module
   - Added tests for basic methods (x_flip, z_flip, meas_flip, children, parents)
   - Added tests for Pauli axis cache initialization and chain cache memoization

docs/source/command.rst

@@ -24,6 +24,9 @@ Command Classes
 .. autoclass:: graphqomb.command.Z
     :members:
 
+.. autoclass:: graphqomb.command.TICK
+    :members:
+
 Type Alias
 ----------
 

examples/entanglement_scheduling_demo.py

@@ -0,0 +1,115 @@
+"""Entanglement Scheduling and TICK Command Demo
+================================================
+
+This example demonstrates the new entanglement scheduling functionality
+and TICK commands in graphqomb.
+"""
+
+from graphqomb.common import Plane, PlannerMeasBasis
+from graphqomb.graphstate import GraphState
+from graphqomb.pattern import print_pattern
+from graphqomb.qompiler import qompile
+from graphqomb.schedule_solver import ScheduleConfig, Strategy
+from graphqomb.scheduler import Scheduler
+
+# Create a simple graph state
+print("=== Entanglement Scheduling Demo ===\n")
+print("1. Creating graph state...")
+node_labels = ["input", "middle1", "middle2", "output"]
+edges = [("input", "middle1"), ("middle1", "middle2"), ("middle2", "output")]
+inputs = ["input"]
+outputs = ["output"]
+meas_bases = {
+    "input": PlannerMeasBasis(Plane.XY, 0.0),
+    "middle1": PlannerMeasBasis(Plane.XY, 0.0),
+    "middle2": PlannerMeasBasis(Plane.XY, 0.0),
+}
+
+graph, node_map = GraphState.from_graph(
+    nodes=node_labels,
+    edges=edges,
+    inputs=inputs,
+    outputs=outputs,
+    meas_bases=meas_bases,
+)
+
+node0 = node_map["input"]
+node1 = node_map["middle1"]
+node2 = node_map["middle2"]
+node3 = node_map["output"]
+
+print(f"   Nodes: {list(graph.physical_nodes)}")
+print(f"   Input: {list(graph.input_node_indices.keys())}")
+print(f"   Output: {list(graph.output_node_indices.keys())}")
+print(f"   Edges: {list(graph.physical_edges)}")
+
+# Define flow
+flow = {node0: {node1}, node1: {node2}, node2: {node3}}
+
+# Create scheduler
+print("\n2. Creating scheduler and solving schedule...")
+scheduler = Scheduler(graph, flow)
+config = ScheduleConfig(strategy=Strategy.MINIMIZE_SPACE)
+success = scheduler.solve_schedule(config)
+
+if success:
+    print("   Scheduling successful!")
+    print(f"   Number of time slices: {scheduler.num_slices()}")
+
+    # Show preparation times
+    prep_times = {k: v for k, v in scheduler.prepare_time.items() if v is not None}
+    print(f"   Preparation times: {prep_times}")
+
+    # Show measurement times
+    meas_times = {k: v for k, v in scheduler.measure_time.items() if v is not None}
+    print(f"   Measurement times: {meas_times}")
+
+    # Show entanglement times (auto-scheduled by solve_schedule)
+    print("\n3. Entanglement times (auto-scheduled)...")
+    ent_times = {edge: time for edge, time in scheduler.entangle_time.items() if time is not None}
+    print(f"   Entanglement times: {ent_times}")
+
+    # Show detailed timeline
+    print("\n4. Detailed timeline (Prep, Entangle, Measure):")
+    timeline = scheduler.timeline
+    for time_idx, (prep_nodes, ent_edges, meas_nodes) in enumerate(timeline):
+        print(f"   Time {time_idx}:")
+        if prep_nodes:
+            print(f"     Prepare: {sorted(prep_nodes)}")
+        if ent_edges:
+            edges_str = [f"({min(e)},{max(e)})" for e in ent_edges]
+            print(f"     Entangle: {', '.join(edges_str)}")
+        if meas_nodes:
+            print(f"     Measure: {sorted(meas_nodes)}")
+
+    # Compile pattern (TICK commands are inserted automatically per time slice)
+    print("\n5. Compiling pattern with scheduler-driven TICK commands...")
+    pattern = qompile(graph, flow, scheduler=scheduler)
+    print(f"   Pattern has {len(pattern.commands)} commands")
+    print(f"   Maximum space usage: {pattern.max_space} qubits")
+
+    print("\n   Pattern commands:")
+    print_pattern(pattern, lim=30)
+
+    # Manual entanglement scheduling example
+    print("\n6. Manual entanglement scheduling example...")
+    scheduler2 = Scheduler(graph, flow)
+    scheduler2.manual_schedule(
+        prepare_time={node1: 0, node2: 1, node3: 2},
+        measure_time={node0: 1, node1: 2, node2: 3},
+        entangle_time={
+            (node0, node1): 0,
+            (node1, node2): 1,
+            (node2, node3): 2,
+        },
+    )
+
+    scheduler2.validate_schedule()  # Will raise ValueError if invalid
+    print("   Manual schedule is valid: True")
+    print(f"   Number of time slices: {scheduler2.num_slices()}")
+
+    pattern_manual = qompile(graph, flow, scheduler=scheduler2)
+    print(f"   Pattern has {len(pattern_manual.commands)} commands")
+    print(f"   Maximum space usage: {pattern_manual.max_space} qubits")
+
+print("\nDemo completed successfully!")

examples/scheduler_pattern_demo.py

@@ -33,7 +33,8 @@
 success_space = scheduler_space.solve_schedule(space_config, timeout=10)
 pattern_space = None
 
-if success_space and scheduler_space.validate_schedule():
+if success_space:
+    scheduler_space.validate_schedule()  # Will raise ValueError if invalid
     print("   Scheduling successful!")
     print(f"   Number of time slices: {scheduler_space.num_slices()}")
 
@@ -66,7 +67,8 @@
 success_time = scheduler_time.solve_schedule(time_config, timeout=10)
 pattern_time = None
 
-if success_time and scheduler_time.validate_schedule():
+if success_time:
+    scheduler_time.validate_schedule()  # Will raise ValueError if invalid
     print("   Scheduling successful!")
     print(f"   Number of time slices: {scheduler_time.num_slices()}")
 
@@ -106,7 +108,8 @@
 success_constrained = scheduler_constrained.solve_schedule(constrained_config, timeout=15)
 pattern_constrained = None
 
-if success_constrained and scheduler_constrained.validate_schedule():
+if success_constrained:
+    scheduler_constrained.validate_schedule()  # Will raise ValueError if invalid
     print("   Scheduling successful!")
     print(f"   Number of time slices: {scheduler_constrained.num_slices()}")
     print(f"   Max qubits constraint: {max_qubits}")
@@ -148,7 +151,8 @@
 scheduler_custom = Scheduler(graph, xflow)
 success_custom = scheduler_custom.solve_schedule(custom_time_config, timeout=10)
 
-if success_custom and scheduler_custom.validate_schedule():
+if success_custom:
+    scheduler_custom.validate_schedule()  # Will raise ValueError if invalid
     print("   Scheduling with custom max_time successful!")
     print(f"   Number of time slices: {scheduler_custom.num_slices()}")
     print(f"   Custom max_time: {custom_time_config.max_time}")

graphqomb/command.py

@@ -7,6 +7,7 @@
 - `M`: Measurement command.
 - `X`: X correction command.
 - `Z`: Z correction command.
+- `TICK`: Time slice separator command.
 - `Command`: Type alias of all commands.
 """
 
@@ -104,7 +105,19 @@ def __str__(self) -> str:
         return f"Z: node={self.node}"
 
 
+@dataclasses.dataclass
+class TICK:
+    """Time slice separator command.
+
+    Marks the boundary between time slices. Commands between two consecutive
+    TICK commands can be executed in parallel within the same time slice.
+    """
+
+    def __str__(self) -> str:
+        return "TICK"
+
+
 if sys.version_info >= (3, 10):
-    Command = N | E | M | X | Z
+    Command = N | E | M | X | Z | TICK
 else:
-    Command = Union[N, E, M, X, Z]
+    Command = Union[N, E, M, X, Z, TICK]

graphqomb/pattern.py

@@ -15,7 +15,7 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
-from graphqomb.command import Command, E, M, N, X, Z
+from graphqomb.command import TICK, Command, E, M, N, X, Z
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -83,12 +83,24 @@ def space(self) -> list[int]:
         for cmd in self.commands:
             if isinstance(cmd, N):
                 nodes += 1
-                space_list.append(nodes)
             elif isinstance(cmd, M):
                 nodes -= 1
+            elif isinstance(cmd, TICK):
+                # TICK does not change the number of qubits
                 space_list.append(nodes)
         return space_list
 
+    @property
+    def depth(self) -> int:
+        """Depth of the pattern (number of TICK commands).
+
+        Returns
+        -------
+        `int`
+            Depth of the pattern
+        """
+        return sum(1 for cmd in self.commands if isinstance(cmd, TICK))
+
 
 def is_runnable(pattern: Pattern) -> None:
     """Check if the pattern is runnable.
@@ -158,6 +170,9 @@ def _ensure_no_operations_on_measured_qubits(pattern: Pattern) -> None:
             if cmd.node in measured:
                 msg = f"Operation on a measured qubit: {cmd}"
                 raise ValueError(msg)
+        elif isinstance(cmd, TICK):
+            # TICK is a time separator and does not operate on qubits
+            pass
         else:
             msg = f"Unknown command kind: {type(cmd)}"
             raise TypeError(msg)
@@ -187,6 +202,9 @@ def _ensure_no_unprepared_qubit_operations(pattern: Pattern) -> None:
         elif isinstance(cmd, (M, X, Z)) and cmd.node not in prepared:
             msg = f"Operation on a qubit that hasn't been prepared yet: {cmd}"
             raise ValueError(msg)
+        elif isinstance(cmd, TICK):
+            # TICK is a time separator and does not operate on qubits
+            pass
 
 
 def _ensure_measurement_consistency(pattern: Pattern) -> None:
@@ -241,7 +259,7 @@ def print_pattern(
     """
 
     def identity_filter(cmd: Command) -> Command | None:
-        return cmd if isinstance(cmd, (N, E, M, X, Z)) else None
+        return cmd if isinstance(cmd, (N, E, M, X, Z, TICK)) else None
 
     if cmd_filter is None:
         cmd_filter = identity_filter

graphqomb/qompiler.py

@@ -12,18 +12,18 @@
 from graphlib import TopologicalSorter
 from typing import TYPE_CHECKING
 
-from graphqomb.command import Command, E, M, N, X, Z
+from graphqomb.command import TICK, Command, E, M, N, X, Z
 from graphqomb.feedforward import check_flow, dag_from_flow
 from graphqomb.graphstate import odd_neighbors
 from graphqomb.pattern import Pattern
 from graphqomb.pauli_frame import PauliFrame
+from graphqomb.scheduler import Scheduler
 
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
     from collections.abc import Set as AbstractSet
 
     from graphqomb.graphstate import BaseGraphState
-    from graphqomb.scheduler import Scheduler
 
 
 def qompile(
@@ -94,31 +94,30 @@ def _qompile(
         compiled pattern
     """
     meas_bases = graph.meas_bases
-    non_input_nodes = graph.physical_nodes - set(graph.input_node_indices)
 
     dag = dag_from_flow(graph, xflow=pauli_frame.xflow, zflow=pauli_frame.zflow)
     topo_order = list(TopologicalSorter(dag).static_order())
     topo_order.reverse()  # children first
 
     commands: list[Command] = []
     if not scheduler:
-        commands.extend(N(node=node) for node in non_input_nodes)
-        commands.extend(E(nodes=edge) for edge in graph.physical_edges)
-        commands.extend(M(node, meas_bases[node]) for node in topo_order if node not in graph.output_node_indices)
-    else:
-        timeline = scheduler.timeline
-        prepared_edges: set[frozenset[int]] = set()
-
-        for time in range(scheduler.num_slices()):
-            prepare_nodes, measure_nodes = timeline[time]
-            for node in measure_nodes:
-                for neighbor in graph.neighbors(node):
-                    edge = frozenset({node, neighbor})
-                    if edge not in prepared_edges:
-                        commands.append(E(nodes=(node, neighbor)))
-                        prepared_edges.add(edge)
-            commands.extend(M(node, meas_bases[node]) for node in measure_nodes)
-            commands.extend(N(node) for node in prepare_nodes)
+        scheduler = Scheduler(graph, pauli_frame.xflow, pauli_frame.zflow)
+        scheduler.solve_schedule()
+
+    timeline = scheduler.timeline
+
+    for time_idx in range(scheduler.num_slices()):
+        prepare_nodes, entangle_edges, measure_nodes = timeline[time_idx]
+
+        # Order within time slice: N -> E -> M
+        commands.extend(N(node) for node in prepare_nodes)
+        for edge in entangle_edges:
+            a, b = edge
+            commands.append(E(nodes=(a, b)))
+        commands.extend(M(node, meas_bases[node]) for node in measure_nodes)
+
+        # Insert TICK between time slices
+        commands.append(TICK())
 
     for node in graph.output_node_indices:
         if meas_basis := graph.meas_bases.get(node):