Merge branch 'master' of github.com:tqsd/EQSN_python into homework_check_eqsn

Author: benjione

eqsn/gates.py

@@ -2,7 +2,7 @@
 import logging
 import numpy as np
 from eqsn.qubit_thread import SINGLE_GATE, MERGE_SEND, MERGE_ACCEPT, MEASURE,\
-                MEASURE_NON_DESTRUCTIVE, GIVE_STATEVECTOR, \
+                MEASURE_NON_DESTRUCTIVE, GIVE_STATEVECTOR, DOUBLE_GATE, \
                 CONTROLLED_GATE, NEW_QUBIT, ADD_MERGED_QUBITS_TO_DICT
 from eqsn.shared_dict import SharedDict
 from eqsn.worker_process import WorkerProcess
@@ -151,6 +151,13 @@ def RZ_gate(self, q_id, rad):
         q = self.shared_dict.get_queues_for_ids([q_id])[0]
         q.put([SINGLE_GATE, x, q_id])
 
+    def custom_gate(self, q_id, gate):
+        """
+        Applies a custom gate to the qubit with q_id.
+        """
+        q = self.shared_dict.get_queues_for_ids([q_id])[0]
+        q.put([SINGLE_GATE, gate, q_id])
+
     def merge_qubits(self, q_id1, q_id2):
         """
         Merges two qubits to one process, if they are not already
@@ -210,6 +217,16 @@ def give_statevector_for(self, q_id):
         qubits, vector = ret.get()
         return qubits, vector
 
+    def custom_two_qubit_gate(self, q_id1, q_id2, gate):
+        self.merge_qubits(q_id1, q_id2)
+        q = self.shared_dict.get_queues_for_ids([q_id1])[0]
+        q.put([DOUBLE_GATE, gate, q_id1, q_id2])
+
+    def custom_controlled_gate(self, applied_to_id, controlled_by_id, gate):
+        self.merge_qubits(applied_to_id, controlled_by_id)
+        q = self.shared_dict.get_queues_for_ids([applied_to_id])[0]
+        q.put([CONTROLLED_GATE, gate, applied_to_id, controlled_by_id])
+
     def measure(self, q_id, non_destructive=False):
         """
         Measures a qubit with an id. If non_destructive is False, the qubit

eqsn/qubit_thread.py

@@ -14,6 +14,7 @@
 NEW_QUBIT = 8
 ADD_MERGED_QUBITS_TO_DICT = 9
 GIVE_STATEVECTOR = 10
+DOUBLE_GATE = 11
 
 
 class QubitThread(object):
@@ -127,6 +128,48 @@ def merge_send(self, channel, chanel2):
         chanel2.put(dp(self.qubits))
         return
 
+    def swap_qubits(self, q_id1, q_id2):
+        """
+        Swaps the position of qubit q_id1 with q_id2
+        in the statevector.
+        """
+        def cnot(q_id1, q_id2):
+            mat = np.asarray([[0, 1], [1, 0]])
+            self.apply_controlled_gate(mat, q_id1, q_id2)
+
+        # Check if they are the same ids
+        if q_id1 == q_id2:
+            return
+
+        # Perform swap
+        cnot(q_id1, q_id2)
+        cnot(q_id2, q_id1)
+        cnot(q_id1, q_id2)
+        # Change ordering in the list
+        i1 = self.qubits.index(q_id1)
+        i2 = self.qubits.index(q_id2)
+        self.qubits[i1], self.qubits[i2] = self.qubits[i2], self.qubits[i1]
+
+    def apply_two_qubit_gate(self, mat, q_id1, q_id2):
+        # Bring the qubits in the right order
+        i2 = self.qubits.index(q_id2)
+        if i2 > 0:
+            new_i1 = i2-1
+            self.swap_qubits(q_id1, self.qubits[new_i1])
+        else:
+            self.swap_qubits(q_id1, self.qubits[0])
+            self.swap_qubits(q_id2, self.qubits[1])
+        apply_mat = mat
+        nr1 = self.qubits.index(q_id1)
+        total_amount = len(self.qubits)
+        before = nr1
+        after = total_amount - nr1 - 2
+        if before > 0:
+            apply_mat = np.kron(np.eye(2 ** before), apply_mat)
+        if after > 0:
+            apply_mat = np.kron(apply_mat, np.eye(2 ** after))
+        self.qubit = np.dot(apply_mat, self.qubit)
+
     def measure_non_destructive(self, q_id, ret_channel):
         """
         Perform a non destructive measurement on qubit with the id.
@@ -241,5 +284,7 @@ def run(self):
                 self.measure_non_destructive(item[1], item[2])
             elif item[0] == GIVE_STATEVECTOR:
                 self.give_statevector(item[1])
+            elif item[0] == DOUBLE_GATE:
+                self.apply_two_qubit_gate(item[1], item[2], item[3])
             else:
                 raise ValueError("Command does not exist!")

eqsn/worker_process.py

@@ -5,7 +5,8 @@
 from eqsn.shared_dict import SharedDict
 from eqsn.qubit_thread import SINGLE_GATE, MERGE_SEND, MERGE_ACCEPT, MEASURE,\
                 MEASURE_NON_DESTRUCTIVE, GIVE_STATEVECTOR, \
-                CONTROLLED_GATE, NEW_QUBIT, ADD_MERGED_QUBITS_TO_DICT, QubitThread
+                CONTROLLED_GATE, NEW_QUBIT, ADD_MERGED_QUBITS_TO_DICT, \
+                DOUBLE_GATE, QubitThread
 
 
 class WorkerProcess(object):
@@ -44,6 +45,8 @@ def run(self):
                 self.add_merged_qubits_to_thread(item[1], item[2])
             elif item[0] == GIVE_STATEVECTOR:
                 self.give_statevector_for(item[1], item[2])
+            elif item[0] == DOUBLE_GATE:
+                self.apply_two_qubit_gate(item[1], item[2], item[3])
             else:
                 raise ValueError("Command does not exist!")
 
@@ -86,6 +89,11 @@ def stop_all(self):
         self.shared_dict.stop_all_threads()
         self.shared_dict.stop_shared_dict()
 
+    def apply_two_qubit_gate(self, gate, q_id1, q_id2):
+        self.merge_qubits(q_id1, q_id2)
+        q = self.shared_dict.get_queues_for_ids([q_id1])[0]
+        q.put([DOUBLE_GATE, gate, q_id1, q_id2])
+
     def apply_single_gate(self, gate, q_id):
         """
         Applies a single gate to a Qubit.

tests/test_custom_gates.py

@@ -0,0 +1,31 @@
+from eqsn import EQSN
+import numpy as np
+
+def test_custom_single_gate():
+    eqsn = EQSN()
+    gate = (1/2)**0.5 * np.array([[1, 1],[1,-1]], dtype=np.csingle)
+    eqsn.new_qubit('1')
+    eqsn.custom_gate('1', gate)
+    eqsn.custom_gate('1', gate)
+    res = eqsn.measure('1')
+    eqsn.stop_all()
+    assert res == 0
+
+def test_custom_controlled_gate():
+    eqsn = EQSN()
+    gate = np.array([[0, 1],[1,0]], dtype=np.csingle)
+    eqsn.new_qubit('1')
+    eqsn.new_qubit('2')
+    eqsn.H_gate('1')
+    eqsn.custom_controlled_gate('2', '1', gate)
+    res1 = eqsn.measure('1')
+    res2 = eqsn.measure('2')
+    eqsn.stop_all()
+    assert res1 == res2
+
+
+if __name__ == "__main__":
+    print("Start custom gate tests...")
+    test_custom_single_gate()
+    test_custom_controlled_gate()
+    print("Finished custom gates test.")

tests/test_custom_two_qubit_gate.py

@@ -0,0 +1,23 @@
+from eqsn import EQSN
+import time
+import numpy as np
+
+
+def test_two_qubit_gate():
+    eqsn = EQSN()
+    eqsn.new_qubit("1")
+    eqsn.new_qubit("2")
+    eqsn.X_gate("1")
+    custom_gate = np.asarray(
+        [[1, 0, 0, 0], [0, 0, 1, 0], [0, 1, 0, 0], [0, 0, 0, 1]]
+    )
+    eqsn.custom_two_qubit_gate("1", "2", custom_gate)
+    m1 = eqsn.measure("1")
+    m2 = eqsn.measure("2")
+    assert m1 == 0
+    assert m2 == 1
+    eqsn.stop_all()
+
+
+if __name__ == "__main__":
+    test_two_qubit_gate()