feat(samples/contrib/modular_exponentiation_quantum_circuit)

samples/contrib/modular_exponentiation_quantum_circuit/QPE.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+#
+# Copyright 2021 The Kubeflow Authors
+#
+# 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.
+
+set -ex
+
+pushd ./backend/src/v2/test
+
+python3 -m pip install --upgrade pip
+python3 -m pip install -r ./requirements-sample-test.txt
+
+popd
+
+# The -u flag makes python output unbuffered, so that we can see real time log.
+# Reference: https://stackoverflow.com/a/107717
+python3 -u ./samples/contrib/QPE_pipeline_example/QPE.py

samples/contrib/modular_exponentiation_quantum_circuit/README.md

@@ -0,0 +1,17 @@
+# Modular Exponentiation Pipeline
+
+This repository contains a Kubeflow pipeline for executing modular exponentiation using Qiskit. The pipeline is implemented using the `kfp` library and can be compiled into a YAML file for deployment on a Kubeflow Pipelines instance.
+
+## Requirements
+
+To run this pipeline, ensure you have the following installed:
+
+- Python 3.11.7
+- Kubeflow Pipelines SDK (`kfp`)
+
+## Pipeline Components
+modular_exponentiation Component
+This component performs modular exponentiation for a given power number using Qiskit. It returns the quantum circuit diagrams for controlled-U operations for specific values of `a` mod 15.
+
+## modular_exponentiation_pipeline Pipeline
+The pipeline defines a single task, modular_exponentiation_task, which executes the modular_exponentiation component with an integer input num.

samples/contrib/modular_exponentiation_quantum_circuit/mod_exp.py

@@ -0,0 +1,49 @@
+from kfp import dsl
+@dsl.component(base_image="python:3.11.7", packages_to_install=['qiskit','matplotlib','pylatexenc'])
+def modular_exponentiation(power_num: int) -> dsl.Artifact:
+    from qiskit import QuantumRegister, QuantumCircuit
+    def mod_func(a, power, show=False):
+
+        assert a in [2,4,7,8,11,13], 'Invalid value of argument a:' + str(a)
+        qrt = QuantumRegister(4, 'target')
+        U = QuantumCircuit(qrt)        
+        for i in range(power):
+            if a in [2,13]:
+                U.swap(0,1)
+                U.swap(1,2)
+                U.swap(2,3)
+            if a in [7,8]:
+                U.swap(2,3)
+                U.swap(1,2)
+                U.swap(0,1)
+            if a in [4, 11]:
+                U.swap(1,3)
+                U.swap(0,2)
+            if a in [7,11,13]:
+                for j in range(4):
+                    U.x(j)
+        if show:
+            print('Below is the circuit of U of ' + f'"{a}^{power} mod 15":') 
+        U = U.to_gate()
+        U.name = f'{a}^{power} mod 15'
+        C_U = U.control()
+        return C_U
+
+    k = []
+    power_arg = power_num
+    for a_arg in [2,4,7,8,11,13]:
+        qrc = QuantumRegister(1, 'control')
+        qrt = QuantumRegister(4, 'target')
+        qc = QuantumCircuit(qrc, qrt)
+        qc.append(mod_func(a_arg, power_arg, show=True), [0,1,2,3,4])
+        print('Below is the circuit of controlled U of ' + f'"{a_arg}^{power_arg} mod 15":')
+        k.append(qc.draw('mpl'))
+    return k
+@dsl.pipeline
+def modular_exponentiation_pipeline(num: int):
+
+    modular_exponentiation_task = modular_exponentiation(power_num=num)
+
+from kfp import compiler
+
+compiler.Compiler().compile(modular_exponentiation_pipeline, 'modular_exponentiation_pipeline.yaml')

samples/contrib/modular_exponentiation_quantum_circuit/sample.py

@@ -0,0 +1,49 @@
+import os
+import unittest
+from dataclasses import dataclass
+from pprint import pprint
+from typing import List
+
+import kfp
+from kfp.dsl.graph_component import GraphComponent
+import QPE
+
+_MINUTE = 60  # seconds
+_DEFAULT_TIMEOUT = 5 * _MINUTE
+
+
+@dataclass
+class TestCase:
+    pipeline_func: GraphComponent
+    timeout: int = _DEFAULT_TIMEOUT
+
+
+class SampleTest(unittest.TestCase):
+    _kfp_host_and_port = os.getenv('KFP_API_HOST_AND_PORT', 'http://localhost:8888')
+    _kfp_ui_and_port = os.getenv('KFP_UI_HOST_AND_PORT', 'http://localhost:8080')
+    _client = kfp.Client(host=_kfp_host_and_port, ui_host=_kfp_ui_and_port)
+
+    def test(self):
+        # 只測試 hello_world 範例
+        test_case = TestCase(pipeline_func=QPE.qpe_pipeline)
+
+        # 直接執行測試案例
+        self.run_test_case(test_case.pipeline_func, test_case.timeout)
+
+    def run_test_case(self, pipeline_func: GraphComponent, timeout: int):
+        # 執行指定的管道函數，並等待完成
+        with self.subTest(pipeline=pipeline_func, msg=pipeline_func.name):
+            run_result = self._client.create_run_from_pipeline_func(pipeline_func=pipeline_func)
+            run_response = run_result.wait_for_run_completion(timeout)
+
+            # 打印運行的詳細信息
+            pprint(run_response.run_details)
+            print("Run details page URL:")
+            print(f"{self._kfp_ui_and_port}/#/runs/details/{run_response.run_id}")
+
+            # 檢查運行狀態是否成功
+            self.assertEqual(run_response.state, "SUCCEEDED")
+
+
+if __name__ == '__main__':
+    unittest.main()