From 8e1ff110f59dc8a21d542dbc3f02ff5c59ed094a Mon Sep 17 00:00:00 2001 From: jordivallsq <151619025+jordivallsq@users.noreply.github.com> Date: Wed, 6 Nov 2024 14:43:15 +0100 Subject: [PATCH 1/4] added timeout and modified set and get for qm (#826) * added timeout and modified set and get for qm * changelog * removed unnecessary elses * removed changes in bus * changed bus name for identifier * added timeout to tests * replaced bus for identifier * TESTS * corrected tests --- docs/releases/changelog-dev.md | 14 ++++++++ .../quantum_machines_cluster.py | 11 +++--- tests/data.py | 20 +++++------ .../test_quantum_machines_cluster.py | 34 +++++++++---------- 4 files changed, 47 insertions(+), 32 deletions(-) diff --git a/docs/releases/changelog-dev.md b/docs/releases/changelog-dev.md index e5bfb6190..bf4714781 100644 --- a/docs/releases/changelog-dev.md +++ b/docs/releases/changelog-dev.md @@ -85,6 +85,20 @@ [#816](https://github.com/qilimanjaro-tech/qililab/pull/816) +- Added a timeout inside quantum machines to control the `wait_for_all_values` function. The timeout is controlled through the runcard as shown in the example: + +``` +instruments: + - name: quantum_machines_cluster + alias: QMM + ... + timeout: 10000 # optional timeout in seconds + octaves: + ... +``` + + [#826](https://github.com/qilimanjaro-tech/qililab/pull/826) + ### Improvements - Legacy linting and formatting tools such as pylint, flake8, isort, bandit, and black have been removed. These have been replaced with Ruff, a more efficient tool that handles both linting and formatting. All configuration settings have been consolidated into the `pyproject.toml` file, simplifying the project's configuration and maintenance. Integration config files like `pre-commit-config.yaml` and `.github/workflows/code_quality.yml` have been updated accordingly. Several rules from Ruff have also been implemented to improve code consistency and quality across the codebase. Additionally, the development dependencies in `dev-requirements.txt` have been updated to their latest versions, ensuring better compatibility and performance. [#813](https://github.com/qilimanjaro-tech/qililab/pull/813) diff --git a/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py b/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py index c1c7ec894..a452cb959 100644 --- a/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py +++ b/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py @@ -68,6 +68,7 @@ class QuantumMachinesClusterSettings(Instrument.InstrumentSettings): octaves: list[dict[str, Any]] controllers: list[dict[str, Any]] elements: list[dict[str, Any]] + timeout: int | None = None def to_qua_config(self) -> DictQuaConfig: """Creates the Quantum Machines QUA config dictionary. @@ -293,7 +294,7 @@ def _get_elements_and_mixers_config(self) -> tuple: mixers = {} for element in self.elements: - bus_name = element["bus"] + bus_name = element["identifier"] element_dict: dict = {"operations": {}} # Flux bus @@ -583,7 +584,7 @@ def set_parameter(self, parameter: Parameter, value: ParameterValue, channel_id: ParameterNotFound: Raised if parameter does not exist. """ bus = str(channel_id) - element = next((element for element in self.settings.elements if element["bus"] == bus), None) + element = next((element for element in self.settings.elements if element["identifier"] == bus), None) if element is None: raise ValueError(f"Bus {bus} was not found in {self.name} settings.") @@ -770,7 +771,7 @@ def get_parameter(self, parameter: Parameter, channel_id: ChannelID | None = Non bus = str(channel_id) settings_config_dict = self.settings.to_qua_config() config_keys = settings_config_dict["elements"][bus] - element = next((element for element in self.settings.elements if element["bus"] == bus), None) + element = next((element for element in self.settings.elements if element["identifier"] == bus), None) if parameter == Parameter.LO_FREQUENCY: if "mixInputs" in config_keys: @@ -805,7 +806,7 @@ def get_parameter(self, parameter: Parameter, channel_id: ChannelID | None = Non return settings_config_dict["elements"][bus]["smearing"] if parameter in [Parameter.THRESHOLD_ROTATION, Parameter.THRESHOLD]: - element = next((element for element in self.settings.elements if element["bus"] == bus), None) + element = next((element for element in self.settings.elements if element["identifier"] == bus), None) if parameter == Parameter.THRESHOLD_ROTATION: return element.get("threshold_rotation", None) # type: ignore if parameter == Parameter.THRESHOLD: @@ -907,7 +908,7 @@ def get_acquisitions(self, job: QmJob | JobApi) -> dict[str, np.ndarray]: QuantumMachinesResult: Quantum Machines result instance. """ result_handles_fetchers = job.result_handles - result_handles_fetchers.wait_for_all_values() + result_handles_fetchers.wait_for_all_values(timeout=self.settings.timeout) results = { name: handle.fetch_all(flat_struct=True) for name, handle in job.result_handles if handle is not None } diff --git a/tests/data.py b/tests/data.py index 0130771c6..355f5d1e4 100644 --- a/tests/data.py +++ b/tests/data.py @@ -837,7 +837,7 @@ class SauronQuantumMachines: "octaves": [], "elements": [ { - "bus": "drive_q0", + "identifier": "drive_q0", "mix_inputs": { "I": {"controller": "con1", "port": 1}, "Q": {"controller": "con1", "port": 2}, @@ -847,7 +847,7 @@ class SauronQuantumMachines: "intermediate_frequency": 6e9, }, { - "bus": "readout_q0", + "identifier": "readout_q0", "mix_inputs": { "I": {"controller": "con1", "port": 3}, "Q": {"controller": "con1", "port": 4}, @@ -861,7 +861,7 @@ class SauronQuantumMachines: "threshold": 0.09, "intermediate_frequency": 6e9, }, - {"bus": "flux_q0", "single_input": {"controller": "con1", "port": 5}}, + {"identifier": "flux_q0", "single_input": {"controller": "con1", "port": 5}}, ], "run_octave_calibration": False, } @@ -908,14 +908,14 @@ class SauronQuantumMachines: ], "elements": [ { - "bus": "drive_q0_rf", + "identifier": "drive_q0_rf", "rf_inputs": {"octave": "octave1", "port": 1}, "digital_inputs": {"controller": "con1", "port": 1, "delay": 87, "buffer": 15}, "digital_outputs": {"controller": "con1", "port": 1}, "intermediate_frequency": 6e9, }, { - "bus": "readout_q0_rf", + "identifier": "readout_q0_rf", "rf_inputs": {"octave": "octave1", "port": 2}, "digital_inputs": {"controller": "con1", "port": 2, "delay": 87, "buffer": 15}, "rf_outputs": {"octave": "octave1", "port": 1}, @@ -994,14 +994,14 @@ class SauronQuantumMachines: ], "elements": [ { - "bus": "drive_q0_rf", + "identifier": "drive_q0_rf", "rf_inputs": {"octave": "octave1", "port": 1}, "digital_inputs": {"controller": "con1", "port": 1, "delay": 87, "buffer": 15}, "digital_outputs": {"controller": "con1", "port": 1}, "intermediate_frequency": 6e9, }, { - "bus": "readout_q0_rf", + "identifier": "readout_q0_rf", "rf_inputs": {"octave": "octave1", "port": 2}, "digital_inputs": {"controller": "con1", "port": 2, "delay": 87, "buffer": 15}, "rf_outputs": {"octave": "octave1", "port": 1}, @@ -1084,14 +1084,14 @@ class SauronQuantumMachines: ], "elements": [ { - "bus": "drive_q0_rf", + "identifier": "drive_q0_rf", "rf_inputs": {"octave": "octave1", "port": 1}, "digital_inputs": {"controller": "con1", "port": 1, "delay": 87, "buffer": 15}, "digital_outputs": {"controller": "con1", "fem": 1, "port": 1}, "intermediate_frequency": 6e9, }, { - "bus": "readout_q0_rf", + "identifier": "readout_q0_rf", "rf_inputs": {"octave": "octave1", "port": 2}, "digital_inputs": {"controller": "con1", "port": 2, "delay": 87, "buffer": 15}, "rf_outputs": {"octave": "octave1", "port": 1}, @@ -1099,7 +1099,7 @@ class SauronQuantumMachines: "time_of_flight": 40, "smearing": 10, }, - {"bus": "flux_q0", "single_input": {"controller": "con1", "fem": 1, "port": 5}}, + {"identifier": "flux_q0", "single_input": {"controller": "con1", "fem": 1, "port": 5}}, ], "run_octave_calibration": True, } diff --git a/tests/instruments/quantum_machines/test_quantum_machines_cluster.py b/tests/instruments/quantum_machines/test_quantum_machines_cluster.py index 2bdfb0997..cc3110e40 100644 --- a/tests/instruments/quantum_machines/test_quantum_machines_cluster.py +++ b/tests/instruments/quantum_machines/test_quantum_machines_cluster.py @@ -8,14 +8,14 @@ import pytest from qm import Program from qm.qua import play, program +from tests.data import SauronQuantumMachines +from tests.test_utils import build_platform from qililab.instruments.instrument import ParameterNotFound from qililab.instruments.quantum_machines import QuantumMachinesCluster from qililab.platform import Platform from qililab.settings import Settings from qililab.typings import Parameter -from tests.data import SauronQuantumMachines -from tests.test_utils import build_platform @pytest.fixture(name="qua_program") @@ -56,7 +56,7 @@ def fixture_qmm(): "octaves": [], "elements": [ { - "bus": "drive_q0", + "identifier": "drive_q0", "mix_inputs": { "I": {"controller": "con1", "port": 1}, "Q": {"controller": "con1", "port": 2}, @@ -66,7 +66,7 @@ def fixture_qmm(): "intermediate_frequency": 6e9, }, { - "bus": "readout_q0", + "identifier": "readout_q0", "mix_inputs": { "I": {"controller": "con1", "port": 3}, "Q": {"controller": "con1", "port": 4}, @@ -80,7 +80,7 @@ def fixture_qmm(): "threshold": 0.09, "intermediate_frequency": 6e9, }, - {"bus": "flux_q0", "single_input": {"controller": "con1", "port": 5}}, + {"identifier": "flux_q0", "single_input": {"controller": "con1", "port": 5}}, ], "run_octave_calibration": False, } @@ -134,14 +134,14 @@ def fixture_qmm_with_octave(): ], "elements": [ { - "bus": "drive_q0_rf", + "identifier": "drive_q0_rf", "rf_inputs": {"octave": "octave1", "port": 1}, "digital_inputs": {"controller": "con1", "port": 1, "delay": 87, "buffer": 15}, "digital_outputs": {"controller": "con1", "port": 1}, "intermediate_frequency": 6e9, }, { - "bus": "readout_q0_rf", + "identifier": "readout_q0_rf", "rf_inputs": {"octave": "octave1", "port": 2}, "digital_inputs": {"controller": "con1", "port": 2, "delay": 87, "buffer": 15}, "rf_outputs": {"octave": "octave1", "port": 1}, @@ -227,14 +227,14 @@ def fixture_qmm_with_octave_custom_connectivity(): ], "elements": [ { - "bus": "drive_q0_rf", + "identifier": "drive_q0_rf", "rf_inputs": {"octave": "octave1", "port": 1}, "digital_inputs": {"controller": "con1", "port": 1, "delay": 87, "buffer": 15}, "digital_outputs": {"controller": "con1", "port": 1}, "intermediate_frequency": 6e9, }, { - "bus": "readout_q0_rf", + "identifier": "readout_q0_rf", "rf_inputs": {"octave": "octave1", "port": 2}, "digital_inputs": {"controller": "con1", "port": 2, "delay": 87, "buffer": 15}, "rf_outputs": {"octave": "octave1", "port": 1}, @@ -324,14 +324,14 @@ def fixture_qmm_with_opx1000(): ], "elements": [ { - "bus": "drive_q0_rf", + "identifier": "drive_q0_rf", "rf_inputs": {"octave": "octave1", "port": 1}, "digital_inputs": {"controller": "con1", "port": 1, "delay": 87, "buffer": 15}, "digital_outputs": {"controller": "con1", "fem": 1, "port": 1}, "intermediate_frequency": 6e9, }, { - "bus": "readout_q0_rf", + "identifier": "readout_q0_rf", "rf_inputs": {"octave": "octave1", "port": 2}, "digital_inputs": {"controller": "con1", "port": 2, "delay": 87, "buffer": 15}, "rf_outputs": {"octave": "octave1", "port": 1}, @@ -339,7 +339,7 @@ def fixture_qmm_with_opx1000(): "time_of_flight": 40, "smearing": 10, }, - {"bus": "flux_q0", "single_input": {"controller": "con1", "fem": 1, "port": 5}}, + {"identifier": "flux_q0", "single_input": {"controller": "con1", "fem": 1, "port": 5}}, ], "run_octave_calibration": True, } @@ -386,7 +386,7 @@ def __init__(self): self.values = [("I", MockSingleHandle()), ("Q", MockSingleHandle())] self.index = 0 - def wait_for_all_values(self): + def wait_for_all_values(self, timeout: int | None = None): """Mocks waiting for all values method from streamer""" return MagicMock @@ -586,7 +586,7 @@ def test_get_controller_from_element_wrong_key_raises_error( qmm.initial_setup() qmm.turn_on() - element = next((element for element in qmm.settings.elements if element["bus"] == "readout_q0"), None) + element = next((element for element in qmm.settings.elements if element["identifier"] == "readout_q0"), None) with pytest.raises( ValueError, @@ -754,7 +754,7 @@ def test_set_parameter_method( qmm.set_parameter(parameter=parameter, value=value, channel_id=bus) - element = next((element for element in qmm.settings.elements if element["bus"] == bus)) + element = next((element for element in qmm.settings.elements if element["identifier"] == bus)) if parameter == Parameter.IF: assert value == element["intermediate_frequency"] if parameter == Parameter.THRESHOLD_ROTATION: @@ -973,10 +973,10 @@ def test_get_parameter_method( assert value == settings_config_dict["elements"][bus]["smearing"] if parameter == Parameter.THRESHOLD_ROTATION: - element = next((element for element in qmm.settings.elements if element["bus"] == bus)) + element = next((element for element in qmm.settings.elements if element["identifier"] == bus)) assert value == element.get("threshold_rotation", None) if parameter == Parameter.THRESHOLD: - element = next((element for element in qmm.settings.elements if element["bus"] == bus)) + element = next((element for element in qmm.settings.elements if element["identifier"] == bus)) assert value == element.get("threshold", None) if parameter == Parameter.DC_OFFSET: From 07d2eadf0c70eeb62d847951db66a3ff865abb32 Mon Sep 17 00:00:00 2001 From: jordivallsq <151619025+jordivallsq@users.noreply.github.com> Date: Wed, 6 Nov 2024 15:50:12 +0100 Subject: [PATCH 2/4] added if in single input and default as 0 (#807) * added if in single input and default as 0 * changelog --- docs/releases/changelog-dev.md | 4 ++++ .../instruments/quantum_machines/quantum_machines_cluster.py | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/docs/releases/changelog-dev.md b/docs/releases/changelog-dev.md index bf4714781..7185c775e 100644 --- a/docs/releases/changelog-dev.md +++ b/docs/releases/changelog-dev.md @@ -2,6 +2,10 @@ ### New features since last release +- Added intermediate frequency to single input lines on qm. The default is 0 (this prevents some bugs from qua-qm). Now it is possible to use the set_parameter IF and qm.set_frequency for buses with single_input. + +[#807](https://github.com/qilimanjaro-tech/qililab/pull/807) + - A new `GetParameter` operation has been added to the Experiment class, accessible via the `.get_parameter()` method. This allows users to dynamically retrieve parameters during experiment execution, which is particularly useful when some variables do not have a value at the time of experiment definition but are provided later through external operations. The operation returns a `Variable` that can be used seamlessly within `SetParameter` and `ExecuteQProgram`. Example: diff --git a/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py b/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py index a452cb959..c8d5410b8 100644 --- a/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py +++ b/src/qililab/instruments/quantum_machines/quantum_machines_cluster.py @@ -299,6 +299,9 @@ def _get_elements_and_mixers_config(self) -> tuple: # Flux bus if "single_input" in element: + intermediate_frequency = ( + int(element["intermediate_frequency"]) if "intermediate_frequency" in element else 0 + ) element_dict["singleInput"] = { "port": ( ( @@ -310,6 +313,7 @@ def _get_elements_and_mixers_config(self) -> tuple: else (element["single_input"]["controller"], element["single_input"]["port"]) ), } + element_dict["intermediate_frequency"] = intermediate_frequency # IQ bus elif "mix_inputs" in element: mixer_name = f"mixer_{bus_name}" From 6b1e3ecf7c75856e9f1f81772165595465c02fe6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adri=C3=A0=20Blanco=20Culla?= <104149115+ziiiki@users.noreply.github.com> Date: Thu, 7 Nov 2024 16:43:01 +0100 Subject: [PATCH 3/4] modify magic method + unit tests (#828) * modify magic method + unit tests * imporove error message * code quality * update changelog * solve space typo --- docs/releases/changelog-dev.md | 6 +++++- src/qililab/slurm.py | 6 +++++- tests/test_slurm.py | 29 ++++++++++++++++++++--------- 3 files changed, 30 insertions(+), 11 deletions(-) diff --git a/docs/releases/changelog-dev.md b/docs/releases/changelog-dev.md index 7185c775e..243ab42f5 100644 --- a/docs/releases/changelog-dev.md +++ b/docs/releases/changelog-dev.md @@ -2,6 +2,10 @@ ### New features since last release +- Support GRES in %%submit_job magic method + +[#828](https://github.com/qilimanjaro-tech/qililab/pull/828) + - Added intermediate frequency to single input lines on qm. The default is 0 (this prevents some bugs from qua-qm). Now it is possible to use the set_parameter IF and qm.set_frequency for buses with single_input. [#807](https://github.com/qilimanjaro-tech/qililab/pull/807) @@ -101,7 +105,7 @@ instruments: ... ``` - [#826](https://github.com/qilimanjaro-tech/qililab/pull/826) +[#826](https://github.com/qilimanjaro-tech/qililab/pull/826) ### Improvements diff --git a/src/qililab/slurm.py b/src/qililab/slurm.py index 2b942db6a..90f28d853 100644 --- a/src/qililab/slurm.py +++ b/src/qililab/slurm.py @@ -45,6 +45,7 @@ def is_variable_used(code, variable): " the results of the job, you need to call `variable.result()`.", ) @argument("-p", "--partition", help="Name of the partition where you want to execute the SLURM job.") +@argument("-g", "--gres", default=None, help="GRES (chip) where you want to execute the SLURM job.") @argument("-n", "--name", default="submitit", help="Name of the slurm job.") @argument("-t", "--time", default=120, help="Time limit (in minutes) of the job.") @argument( @@ -85,6 +86,7 @@ def submit_job(line: str, cell: str, local_ns: dict) -> None: args = parse_argstring(submit_job, line) output = args.output partition = args.partition + gres = args.gres job_name = args.name time_limit = int(args.time) folder_path = args.logs @@ -92,6 +94,8 @@ def submit_job(line: str, cell: str, local_ns: dict) -> None: begin_time = args.begin low_priority = args.low_priority + if gres is None: + raise ValueError("GRES needs to be provided! See the available ones typing 'sinfo -o '%G'' in the terminal") nice_factor = 0 if low_priority in ["True", "true"]: nice_factor = 1000000 # this ensures Lab jobs have 0 priority, same as QaaS jobs @@ -108,7 +112,7 @@ def submit_job(line: str, cell: str, local_ns: dict) -> None: slurm_partition=partition, name=job_name, timeout_min=time_limit, - slurm_additional_parameters={"begin": begin_time, "nice": nice_factor}, + slurm_additional_parameters={"begin": begin_time, "nice": nice_factor, "gres": f"{gres}:1"}, ) # Compile the code defined above diff --git a/tests/test_slurm.py b/tests/test_slurm.py index d7fec0a37..4dfc76fc4 100644 --- a/tests/test_slurm.py +++ b/tests/test_slurm.py @@ -4,9 +4,8 @@ from unittest.mock import MagicMock, patch import pytest -from IPython.testing.globalipapp import start_ipython - import qililab as ql +from IPython.testing.globalipapp import start_ipython slurm_job_data_test = "slurm_job_data_test" @@ -34,7 +33,7 @@ def test_submit_job(self, ip): ip.run_cell(raw_cell="a=1\nb=1") ip.run_cell_magic( magic_name="submit_job", - line=f"-o results -p debug -l {slurm_job_data_test} -n unit_test -e local", + line=f"-o results -g aQPU1 -l {slurm_job_data_test} -n unit_test -e local", cell="results = a+b ", ) time.sleep(4) @@ -48,7 +47,19 @@ def test_submit_job_output_not_assigned(self, ip): ): ip.run_cell_magic( magic_name="submit_job", - line=f"-o results -p debug -l {slurm_job_data_test} -n unit_test -e local", + line=f"-o results -g aQPU1 -l {slurm_job_data_test} -n unit_test -e local", + cell="a+b", + ) + + def test_submit_job_no_gres_provided(self, ip): + """Check ValueError is raised in case GRES is not provided.""" + ip.run_cell(raw_cell="a=1\nb=1") + with pytest.raises( + ValueError, match="GRES needs to be provided! See the available ones typing 'sinfo -o '%G'' in the terminal" + ): + ip.run_cell_magic( + magic_name="submit_job", + line=f"-o results -l {slurm_job_data_test} -n unit_test -e local", cell="a+b", ) @@ -59,7 +70,7 @@ def test_submit_job_with_random_file_in_logs_folder(self, ip): ) ip.run_cell_magic( magic_name="submit_job", - line=f"-o results -p debug -l {slurm_job_data_test} -n unit_test -e local", + line=f"-o results -g aQPU1 -l {slurm_job_data_test} -n unit_test -e local", cell="results=a+b", ) time.sleep(4) # give time to ensure processes are finished @@ -72,7 +83,7 @@ def test_submit_job_delete_info_from_past_jobs(self, ip): for _ in range(int(ql.slurm.num_files_to_keep / 4)): ip.run_cell_magic( magic_name="submit_job", - line=f"-o results -p debug -l {slurm_job_data_test} -n unit_test -e local", + line=f"-o results -g aQPU1 -l {slurm_job_data_test} -n unit_test -e local", cell="results=a+b", ) time.sleep(2) # give time submitit to create the files @@ -83,7 +94,7 @@ def test_submit_job_delete_info_from_past_jobs(self, ip): ) ip.run_cell_magic( magic_name="submit_job", - line=f"-o results -p debug -l {slurm_job_data_test} -n unit_test -e local", + line=f"-o results -g aQPU1 -l {slurm_job_data_test} -n unit_test -e local", cell="results=a+b", ) time.sleep(2) @@ -102,7 +113,7 @@ def test_setting_parameters(self, ip): with patch("qililab.slurm.os"): ip.run_cell_magic( magic_name="submit_job", - line=f"-o results -p debug -l {slurm_job_data_test} -n unit_test -e local -t 5 --begin now+1hour -lp true", + line=f"-o results -p debug -l {slurm_job_data_test} -n unit_test -e local -t 5 --begin now+1hour -lp true -g aQPU1", cell="results=a+b", ) executor.assert_called_once_with(folder=slurm_job_data_test, cluster="local") @@ -110,5 +121,5 @@ def test_setting_parameters(self, ip): slurm_partition="debug", name="unit_test", timeout_min=5, - slurm_additional_parameters={"begin": "now+1hour", "nice": 1000000}, + slurm_additional_parameters={"begin": "now+1hour", "nice": 1000000, "gres":"aQPU1:1"}, ) From dea01bbb9cb6104a7f255c8ec4c5f8245eede4c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Guillermo=20Abad=20L=C3=B3pez?= <109400222+GuillermoAbadLopez@users.noreply.github.com> Date: Fri, 8 Nov 2024 09:57:14 +0100 Subject: [PATCH 4/4] [QHC-700] Add qibo routing (#821) * major refactor to simplify things * various fixes * various changes * update tests * update tests * fix circuit_transpiler tests * add decorators to instrument methods * delete obsolete Pulsar * refactor qblox_module tests * Add qibo routing * Solving bugs in tranpiler * Update circuit_transpiler.py * Adding in cascade up, the placer and router freedom and the final_layout * Update circuit_transpiler.py * Typo * test_typo and general improvements * Adding trivial topology for PulseSchedule in `compile()` * Solving tests * Solve mapping of qubits in tests * fix instruments tests * fix tests * fix tests * fix tests * fix documentation * fix runcard for calibration tests * delete obsolete files, fix mypy * improve codecov * improve codecov * Creating class CircuitRouter * add changelog entry * add changelog entry * Update transpiler.rst * re-enable mypy in precommit * rename circuit_transpiler module to digital * Update circuit_transpiler.py * delete drivers module * add topology to digital_compilation_settings * update changelog * fix documentation * Update data.py * fix is_awg/is_adc docstrings * Supress errors. for specific one * Improve error handling * Improve error handling * Improving erro handling * Adding routing_iterations, to get best stochastic routing * Improving the best routing finding * Make routing iteration, more readable * improve qblox_compiler tests * improve tests * improve tests * improve QCM/QRM RF modules and relevant tests * improve qblox tests * improve qblox tests * improve tests * improve tests * improve tests * improve tests * improve tests * improve tests * add ignore to codecov * improve tests * Make `_iterate_routing` method private * Solving code quality * Remove duplicated method * Add initialization test * Add basic routing test * Adding unit tests * Improve unit testing * Expand unit tests * Update test_platform.py * Update platform.py * Adding more test, and improve type checking * Improving tests * Addint `transpile_circuits` test * Adding route_circuit test * Improving coverage to 100 * Improve tests * Adding release and Improving docstrings * Update changelog-dev.md * Indent changelog * [QHC-700] Expand transpiler (#823) * Improve transpiler * Update circuit_transpiler.py * Add pairs of hermitian gate cancellation * Making the transpiler more modular * Make transpiler more modular * Update circuit_to_pulses.py * Update circuit_transpiler.py * Update circuit_to_pulses.py * Solve problem with Drag initialization, and tests * Solve tests * Solving and testing circuit optimizer * Solving unit tests * Adding unit test * Solve.unit test * Update changelog-dev.md * Update changelog-dev.md * Improve typings and docstrings * Take optimize parameter in cascade up, to the user calls --------- Co-authored-by: Vyron Vasileiadis --- docs/code/transpiler.rst | 4 +- docs/releases/changelog-dev.md | 75 +++- requirements.txt | 2 +- src/qililab/digital/__init__.py | 11 +- src/qililab/digital/circuit_optimizer.py | 263 ++++++++++++ src/qililab/digital/circuit_router.py | 364 ++++++++++++++++ src/qililab/digital/circuit_to_pulses.py | 275 ++++++++++++ src/qililab/digital/circuit_transpiler.py | 493 +++++++++------------- src/qililab/execute_circuit.py | 46 +- src/qililab/platform/platform.py | 67 ++- tests/digital/test_circuit_optimizer.py | 129 ++++++ tests/digital/test_circuit_router.py | 348 +++++++++++++++ tests/digital/test_circuit_transpiler.py | 119 +++++- tests/platform/test_platform.py | 45 +- 14 files changed, 1880 insertions(+), 361 deletions(-) create mode 100644 src/qililab/digital/circuit_optimizer.py create mode 100644 src/qililab/digital/circuit_router.py create mode 100644 src/qililab/digital/circuit_to_pulses.py create mode 100644 tests/digital/test_circuit_optimizer.py create mode 100644 tests/digital/test_circuit_router.py diff --git a/docs/code/transpiler.rst b/docs/code/transpiler.rst index d2dbedcbc..8e014146c 100644 --- a/docs/code/transpiler.rst +++ b/docs/code/transpiler.rst @@ -1,4 +1,4 @@ -ql.transpiler -=============== +ql.circuit_transpiler +========================= .. automodule:: qililab.digital diff --git a/docs/releases/changelog-dev.md b/docs/releases/changelog-dev.md index 243ab42f5..b5dfd90f2 100644 --- a/docs/releases/changelog-dev.md +++ b/docs/releases/changelog-dev.md @@ -93,17 +93,68 @@ [#816](https://github.com/qilimanjaro-tech/qililab/pull/816) +- Added routing algorithms to `qililab` in function of the platform connectivity. This is done passing `Qibo` own `Routers` and `Placers` classes, + and can be called from different points of the stack. + + The most common way to route, will be automatically through `qililab.execute_circuit.execute()`, or also from `qililab.platform.execute/compile()`. Another way, would be doing the transpilation/routing directly from an instance of the Transpiler, with: `qililab.digital.circuit_transpiler.transpile/route_circuit()` (with this last one, you can route with a different topology from the platform one, if desired, defaults to platform) + + Example: + + ```python + from qibo import gates + from qibo.models import Circuit + from qibo.transpiler.placer import ReverseTraversal, Trivial + from qibo.transpiler.router import Sabre + from qililab import build_platform + from qililab.circuit_transpiler import CircuitTranspiler + + # Create circuit: + c = Circuit(5) + c.add(gates.CNOT(1, 0)) + + ### From execute_circuit: + # With defaults (ReverseTraversal placer and Sabre routing): + probabilities = ql.execute(c, runcard="./runcards/galadriel.yml", placer= Trivial, router = Sabre, routing_iterations: int = 10,) + # Changing the placer to Trivial, and changing the number of iterations: + probabilities = ql.execute(c, runcard="./runcards/galadriel.yml", + + ### From the platform: + # Create platform: + platform = build_platform(runcard="") + # With defaults (ReverseTraversal placer, Sabre routing) + probabilities = platform.execute(c, num_avg: 1000, repetition_duration: 1000) + # With non-defaults, and specifying the router with kwargs: + probabilities = platform.execute(c, num_avg: 1000, repetition_duration: 1000, placer= Trivial, router = (Sabre, {"lookahead": 2}), routing_iterations: int = 20)) + # With a router instance: + router = Sabre(connectivity=None, lookahead=1) # No connectivity needed, since it will be overwritten by the platform's one + probabilities = platform.execute(c, num_avg: 1000, repetition_duration: 1000, placer=Trivial, router=router) + + ### Using the transpiler directly: + ### (If using the routing from this points of the stack, you can route with a different topology from the platform one) + # Create transpiler: + transpiler = CircuitTranspiler(platform) + # Default Transpilation (ReverseTraversal, Sabre and Platform connectivity): + routed_circ, final_layouts = transpiler.route_circuit([c]) + # With Non-Default Trivial placer, specifying the kwargs, for the router, and different coupling_map: + routed_circ, final_layouts = transpiler.route_circuit([c], placer=Trivial, router=(Sabre, {"lookahead": 2}, coupling_map=)) + # Or finally, Routing with a concrete Routing instance: + router = Sabre(connectivity=None, lookahead=1) # No connectivity needed, since it will be overwritten by the specified in the Transpiler: + routed_circ, final_layouts = transpiler.route_circuit([c], placer=Trivial, router=router, coupling_map=) + ``` + +[#821](https://github.com/qilimanjaro-tech/qililab/pull/821) + - Added a timeout inside quantum machines to control the `wait_for_all_values` function. The timeout is controlled through the runcard as shown in the example: -``` -instruments: - - name: quantum_machines_cluster - alias: QMM - ... - timeout: 10000 # optional timeout in seconds - octaves: - ... -``` + ```json + instruments: + - name: quantum_machines_cluster + alias: QMM + ... + timeout: 10000 # optional timeout in seconds + octaves: + ... + ``` [#826](https://github.com/qilimanjaro-tech/qililab/pull/826) @@ -117,7 +168,7 @@ instruments: Example (for Qblox) - ``` + ```json buses: - alias: readout ... @@ -182,6 +233,10 @@ instruments: - Added a `save_plot=True` parameter to the `plotS21()` method of `ExperimentResults`. When enabled (default: True), the plot is automatically saved in the same directory as the experiment results. [#819](https://github.com/qilimanjaro-tech/qililab/pull/819) +- Improved the transpiler, by making it more modular, and adding a `gate_cancellation()` stage before the transpilation to natives, this stage can be skipped, together with the old `optimize_transpilation()`, if the flag `optimize=False` is passed. + +[#823](https://github.com/qilimanjaro-tech/qililab/pull/823) + ### Breaking changes - Renamed the platform's `execute_anneal_program()` method to `execute_annealing_program()` and updated its parameters. The method now expects `preparation_block` and `measurement_block`, which are strings used to retrieve blocks from the `Calibration`. These blocks are inserted before and after the annealing schedule, respectively. diff --git a/requirements.txt b/requirements.txt index 12c0a4edc..15cf2731b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ pandas==1.5.3 -qibo==0.2.8 +qibo==0.2.12 qblox-instruments==0.11.2 qcodes==0.42.0 qcodes_contrib_drivers==0.18.0 diff --git a/src/qililab/digital/__init__.py b/src/qililab/digital/__init__.py index 2912cacf0..e7f263f52 100644 --- a/src/qililab/digital/__init__.py +++ b/src/qililab/digital/__init__.py @@ -15,21 +15,16 @@ """ This module contains all the decomposition and transpilation methods used within qililab. -.. currentmodule:: qililab - Transpilation ~~~~~~~~~~~~~ -.. autosummary:: - :toctree: api - -Gate Decomposition -~~~~~~~~~~~~~~~~~~ - .. currentmodule:: qililab.digital .. autosummary:: :toctree: api + + ~CircuitTranspiler + """ from .circuit_transpiler import CircuitTranspiler diff --git a/src/qililab/digital/circuit_optimizer.py b/src/qililab/digital/circuit_optimizer.py new file mode 100644 index 000000000..38a44535c --- /dev/null +++ b/src/qililab/digital/circuit_optimizer.py @@ -0,0 +1,263 @@ +# Copyright 2023 Qilimanjaro Quantum Tech +# +# 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. + +"""CircuitOptimizer class""" + +from copy import deepcopy + +from qibo import Circuit, gates + +from qililab import digital +from qililab.settings.digital.digital_compilation_settings import DigitalCompilationSettings + +from .native_gates import Drag + + +class CircuitOptimizer: + """Optimizes a circuit, cancelling redundant gates.""" + + def __init__(self, digital_compilation_settings: DigitalCompilationSettings): # type: ignore # ignore typing to avoid importing platform and causing circular imports + self.digital_compilation_settings = digital_compilation_settings + + @classmethod + def run_gate_cancellations(cls, circuit: Circuit) -> Circuit: + """Main method to run the gate cancellations. Currently only consists of cancelling pairs of hermitian gates. + + Can/Might be extended in the future to include more complex gate cancellations. + + Args: + circuit (Circuit): circuit to optimize. + + Returns: + Circuit: optimized circuit. + """ + return cls.cancel_pairs_of_hermitian_gates(circuit) + + @classmethod + def cancel_pairs_of_hermitian_gates(cls, circuit: Circuit) -> Circuit: + """Optimizes circuit by cancelling adjacent hermitian gates. + + Args: + circuit (Circuit): circuit to optimize. + + Returns: + Circuit: optimized circuit. + """ + # Initial and final circuit gates lists, from which to, one by one, after checks, pass non-cancelled gates: + circ_list: list[tuple] = cls._get_circuit_gates(circuit) + + # We want to do the sweep circuit cancelling gates least once always: + previous_circ_list = deepcopy(circ_list) + output_circ_list = cls._sweep_circuit_cancelling_pairs_of_hermitian_gates(circ_list) + + # And then keep iterating, sweeping over the circuit (cancelling gates) each time, until there is full sweep without any cancellations: + while output_circ_list != previous_circ_list: + previous_circ_list = deepcopy(output_circ_list) + output_circ_list = cls._sweep_circuit_cancelling_pairs_of_hermitian_gates(output_circ_list) + + # Create optimized circuit, from the obtained non-cancelled list: + return cls._create_circuit(output_circ_list, circuit.nqubits) + + def optimize_transpilation(self, circuit: Circuit) -> list[gates.Gate]: + """Optimizes transpiled circuit by applying virtual Z gates. + + This is done by moving all RZ to the left of all operators as a single RZ. The corresponding cumulative rotation + from each RZ is carried on as phase in all drag pulses left of the RZ operator. + + Virtual Z gates are also applied to correct phase errors from CZ gates. + + The final RZ operator left to be applied as the last operator in the circuit can afterwards be removed since the last + operation is going to be a measurement, which is performed on the Z basis and is therefore invariant under rotations + around the Z axis. + + This last step can also be seen from the fact that an RZ operator applied on a single qubit, with no operations carried + on afterwards induces a phase rotation. Since phase is an imaginary unitary component, its absolute value will be 1 + independent on any (unitary) operations carried on it. + + Mind that moving an operator to the left is equivalent to applying this operator last so + it is actually moved to the _right_ of ``Circuit.queue`` (last element of list). + + For more information on virtual Z gates, see https://arxiv.org/abs/1612.00858 + + Args: + circuit (Circuit): circuit with native gates, to optimize. + + Returns: + list[gates.Gate] : list of re-ordered gates + """ + nqubits: int = circuit.nqubits + ngates: list[gates.Gate] = circuit.queue + + supported_gates = ["rz", "drag", "cz", "wait", "measure"] + new_gates = [] + shift = dict.fromkeys(range(nqubits), 0) + for gate in ngates: + if gate.name not in supported_gates: + raise NotImplementedError(f"{gate.name} not part of native supported gates {supported_gates}") + if isinstance(gate, gates.RZ): + shift[gate.qubits[0]] += gate.parameters[0] + # add CZ phase correction + elif isinstance(gate, gates.CZ): + gate_settings = self.digital_compilation_settings.get_gate( + name=gate.__class__.__name__, qubits=gate.qubits + ) + control_qubit, target_qubit = gate.qubits + corrections = next( + ( + event.pulse.options + for event in gate_settings + if ( + event.pulse.options is not None + and f"q{control_qubit}_phase_correction" in event.pulse.options + ) + ), + None, + ) + if corrections is not None: + shift[control_qubit] += corrections[f"q{control_qubit}_phase_correction"] + shift[target_qubit] += corrections[f"q{target_qubit}_phase_correction"] + new_gates.append(gate) + else: + # if gate is drag pulse, shift parameters by accumulated Zs + if isinstance(gate, Drag): + # create new drag pulse rather than modify parameters of the old one + gate = Drag(gate.qubits[0], gate.parameters[0], gate.parameters[1] - shift[gate.qubits[0]]) + + # append gate to optimized list + new_gates.append(gate) + + return new_gates + + @staticmethod + def _get_circuit_gates(circuit: Circuit) -> list[tuple]: + """Get the gates of the circuit. + + Args: + circuit (qibo.models.Circuit): Circuit to get the gates from. + + Returns: + list[tuple]: List of gates in the circuit. Where each gate is a tuple of ('name', 'init_args', 'initi_kwargs'). + """ + return [(type(gate).__name__, gate.init_args, gate.init_kwargs) for gate in circuit.queue] + + @staticmethod + def _create_gate(gate_class: str, gate_args: list | int, gate_kwargs: dict) -> gates.Gate: + """Converts a tuple representation of qibo gate (name, qubits) into a Gate object. + + Args: + gate_class (str): The class name of the gate. Can be any Qibo or Qililab supported class. + gate_args (list | int): The qubits the gate acts on. + gate_kwargs (dict): The kwargs of the gate. + + Returns: + gates.Gate: The qibo Gate object. + """ + # Solve Identity gate, argument int issue: + gate_args = [gate_args] if isinstance(gate_args, int) else gate_args + + return ( + getattr(digital, gate_class)(*gate_args, **gate_kwargs) + if gate_class in {"Drag", "Wait"} + else getattr(gates, gate_class)(*gate_args, **gate_kwargs) + ) + + @classmethod + def _create_circuit(cls, gates_list: list[tuple], nqubits: int) -> Circuit: + """Converts a list of gates (name, qubits) into a qibo Circuit object. + + Args: + gates_list (list[tuple]): List of gates in the circuit. Where each gate is a tuple of ('name', 'init_args', 'initi_kwargs') + nqubits (int): Number of qubits in the circuit. + + Returns: + Circuit: The qibo Circuit object. + """ + # Create optimized circuit, from the obtained non-cancelled list: + output_circuit = Circuit(nqubits) + for gate, gate_args, gate_kwargs in gates_list: + qibo_gate = cls._create_gate(gate, gate_args, gate_kwargs) + output_circuit.add(qibo_gate) + + return output_circuit + + @classmethod + def _sweep_circuit_cancelling_pairs_of_hermitian_gates(cls, circ_list: list[tuple]) -> list[tuple]: + """Cancels adjacent gates in a circuit. + + Args: + circ_list (list[tuple]): List of gates in the circuit. Where each gate is a tuple of ('name', 'init_args', 'initi_kwargs') + + Returns: + list[tuple]: List of gates in the circuit, after cancelling adjacent gates. Where each gate is a tuple of ('name', 'init_args', 'initi_kwargs') + """ + # List of gates, that are available for cancellation: + hermitian_gates: list = ["H", "X", "Y", "Z", "CNOT", "CZ", "SWAP"] + + output_circ_list: list[tuple] = [] + + while circ_list: # If original circuit list, is empty or has one gate remaining, we are done: + if len(circ_list) == 1: + output_circ_list.append(circ_list[0]) + break + + # Gate of the original circuit, to find a match for: + gate, gate_args, gate_kwargs = circ_list.pop(0) + gate_qubits = cls._extract_qubits(gate_args) # Assuming qubits are the first two args + + # If gate is not hermitian (can't be cancelled), add it to the output circuit and continue: + if gate not in hermitian_gates: + output_circ_list.append((gate, gate_args, gate_kwargs)) + continue + + subend = False + for i in range(len(circ_list)): + # Next gates, to compare the original with: + comp_gate, comp_args, comp_kwargs = circ_list[i] + comp_qubits = cls._extract_qubits(comp_args) # Assuming qubits are the first two args + + # Simplify duplication, if same gate and qubits found, without any other in between: + if gate == comp_gate and gate_args == comp_args and gate_kwargs == comp_kwargs: + circ_list.pop(i) + break + + # Add gate, if there is no other gate that acts on the same qubits: + if i == len(circ_list) - 1: + output_circ_list.append((gate, gate_args, gate_kwargs)) + break + + # Add gate and leave comparison_gate loop, if we find a gate in common qubit, that prevents contraction: + for gate_qubit in gate_qubits: + if gate_qubit in comp_qubits: + output_circ_list.append((gate, gate_args, gate_kwargs)) + subend = True + break + if subend: + break + + return output_circ_list + + @staticmethod + def _extract_qubits(gate_args: list | int) -> list: + """Extract qubits from gate_args. + + Args: + gate_args (list | int): The arguments of the gate. + + Returns: + list: The qubits of the gate in an iterable. + """ + # Assuming qubits are the first one or two args: + if isinstance(gate_args, int): + return [gate_args] + return gate_args if len(gate_args) <= 2 else gate_args[:2] diff --git a/src/qililab/digital/circuit_router.py b/src/qililab/digital/circuit_router.py new file mode 100644 index 000000000..b901b87a5 --- /dev/null +++ b/src/qililab/digital/circuit_router.py @@ -0,0 +1,364 @@ +# Copyright 2023 Qilimanjaro Quantum Tech +# +# 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. + +"""CircuitRouter class""" + +import contextlib +import re + +import networkx as nx +from qibo import Circuit, gates +from qibo.transpiler.optimizer import Preprocessing +from qibo.transpiler.pipeline import Passes +from qibo.transpiler.placer import Placer, ReverseTraversal, StarConnectivityPlacer +from qibo.transpiler.router import Router, Sabre, StarConnectivityRouter + +from qililab.config import logger + + +class CircuitRouter: + """Handles circuit routing, using a Placer and Router, and a coupling map. It has a single accessible method: + + - ``route(circuit: Circuit) -> tuple[Circuit, dict]``: Routes the virtual/logical qubits of a circuit, to the chip's physical qubits. + + Args: + connectivity (nx.graph): Chip connectivity. + placer (Placer | type[Placer] | tuple[type[Placer], dict], optional): `Placer` instance, or subclass `type[Placer]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `ReverseTraversal`. + router (Router | type[Router] | tuple[type[Router], dict], optional): `Router` instance, or subclass `type[Router]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `Sabre`. + """ + + def __init__( + self, + connectivity: nx.graph, + placer: Placer | type[Placer] | tuple[type[Placer], dict] | None = None, + router: Router | type[Router] | tuple[type[Router], dict] | None = None, + ): + self.connectivity = connectivity + """nx.graph: Chip connectivity.""" + + self.preprocessing = Preprocessing(self.connectivity) + """Preprocessing: Stage to add qubits in the original circuit to match the number of qubits in the chip.""" + + self.router = self._build_router(router, self.connectivity) + """Router: Routing stage, where the final_layout and swaps will be created. Defaults to Sabre.""" + + self.placer = self._build_placer(placer, self.router, self.connectivity) + """Placer: Layout stage, where the initial_layout will be created. Defaults to ReverseTraversal.""" + + # Cannot use Star algorithms for non star-connectivities: + if self._if_star_algorithms_for_nonstar_connectivity(self.connectivity, self.placer, self.router): + raise (ValueError("StarConnectivity Placer and Router can only be used with star topologies")) + + # Transpilation pipeline passes: + self.pipeline = Passes([self.preprocessing, self.placer, self.router], self.connectivity) + """Routing pipeline passes: Preprocessing, Placer and Router passes. Defaults to Passes([Preprocessing, ReverseTraversal, Sabre]).""" + # 1) Preprocessing adds qubits in the original circuit to match the number of qubits in the chip. + # 2) Routing stage, where the final_layout and swaps will be created. + # 3) Layout stage, where the initial_layout will be created. + + def route(self, circuit: Circuit, iterations: int = 10) -> tuple[Circuit, dict[str, int]]: + """Routes the virtual/logical qubits of a circuit, to the chip's physical qubits. + + **Examples:** + + If we instantiate some ``Circuit``, ``Platform`` and ``CircuitTranspiler`` objects like: + + .. code-block:: python + + from qibo import gates + from qibo.models import Circuit + from qibo.transpiler.placer import ReverseTraversal, Trivial + from qibo.transpiler.router import Sabre + from qililab import build_platform + from qililab.circuit_transpiler import CircuitTranspiler + + # Create circuit: + c = Circuit(5) + c.add(gates.CNOT(1, 0)) + + # Create platform: + platform = build_platform(runcard="") + coupling_map = platform.digital_compilation_settings.topology + + # Create transpiler: + transpiler = CircuitTranspiler(platform) + + Now we can transpile like: + + .. code-block:: python + + # Default Transpilation: + routed_circuit, final_layouts = transpiler.route_circuit([c]) # Defaults to ReverseTraversal, Sabre and platform connectivity + + # Non-Default Trivial placer, and coupling_map specified: + routed_circuit, final_layouts = transpiler.route_circuit([c], placer=Trivial, router=Sabre, coupling_map) + + # Specifying one of the a kwargs: + routed_circuit, final_layouts = transpiler.route_circuit([c], placer=Trivial, router=(Sabre, {"lookahead": 2})) + + Args: + circuit (Circuit): circuit to route. + iterations (int, optional): Number of times to repeat the routing pipeline, to keep the best stochastic result. Defaults to 10. + + Returns: + tuple [Circuit, dict[str, int]: routed circuit and final layout of the circuit. + + Raises: + ValueError: If StarConnectivity Placer and Router are used with non-star topologies. + ValueError: If the final layout is not valid, i.e. a qubit is mapped to more than one physical qubit. + """ + # Call the routing pipeline on the circuit, multiple times, and keep the best stochastic result: + best_transp_circ, best_final_layout, least_swaps = self._iterate_routing(self.pipeline, circuit, iterations) + + if self._if_layout_is_not_valid(best_final_layout): + raise ValueError( + f"The final layout: {best_final_layout} is not valid. i.e. a qubit is mapped to more than one physical qubit. Try again, if the problem persists, try another placer/routing algorithm." + ) + + if least_swaps is not None: + logger.info(f"The best found routing, has {least_swaps} swaps.") + else: + logger.info("No routing was done. Most probably due to routing iterations being 0.") + + return best_transp_circ, best_final_layout + + @staticmethod + def _iterate_routing( + routing_pipeline: Passes, circuit: Circuit, iterations: int = 10 + ) -> tuple[Circuit, dict[str, int], int | None]: + """Iterates the routing pipeline, to keep the best stochastic result. + + Args: + routing_pipeline (Passes): Transpilation pipeline passes. + circuit (Circuit): Circuit to route. + iterations (int, optional): Number of times to repeat the routing pipeline, to keep the best stochastic result. Defaults to 10. + + Returns: + tuple[Circuit, dict[str, int], int]: Best transpiled circuit, best final layout and least swaps. + """ + # We repeat the routing pipeline a few times, to keep the best stochastic result: + least_swaps: int | None = None + for _ in range(iterations): + # Call the routing pipeline on the circuit: + transpiled_circ, final_layout = routing_pipeline(circuit) + + # Get the number of swaps in the circuits: + n_swaps = len(transpiled_circ.gates_of_type(gates.SWAP)) + + # Checking which is the best transpilation: + if least_swaps is None or n_swaps < least_swaps: + least_swaps = n_swaps + best_transpiled_circ, best_final_layout = transpiled_circ, final_layout + + # If a mapping needs no swaps, we are finished: + if n_swaps == 0: + break + + return best_transpiled_circ, best_final_layout, least_swaps + + @staticmethod + def _if_star_algorithms_for_nonstar_connectivity(connectivity: nx.Graph, placer: Placer, router: Router) -> bool: + """True if the StarConnectivity Placer or Router are being used without a star connectivity. + + Args: + connectivity (nx.Graph): Chip connectivity. + placer (Placer): Placer instance. + router (Router): Router instance. + + Returns: + bool: True if the StarConnectivity Placer or Router are being used without a star connectivity. + """ + + return not nx.is_isomorphic(connectivity, nx.star_graph(4)) and ( + isinstance(placer, StarConnectivityPlacer) or isinstance(router, StarConnectivityRouter) + ) + + @staticmethod + def _highest_degree_node(connectivity: nx.Graph) -> int: + """Returns the node with the highest degree in the connectivity graph. + + Args: + connectivity (nx.Graph): Chip connectivity. + + Returns: + int: Node with the highest degree in the connectivity graph. + """ + return max(dict(connectivity.degree()).items(), key=lambda x: x[1])[0] + + @staticmethod + def _if_layout_is_not_valid(layout: dict[str, int]) -> bool: + """True if the layout is not valid. + + For example, if a qubit is mapped to more than one physical qubit. Or if the keys or values are not int. + + Args: + layout (dict[str, int]): Initial or final layout of the circuit. + + Returns: + bool: True if the layout is not valid. + """ + return ( + len(layout.values()) != len(set(layout.values())) + or not all(isinstance(value, int) for value in layout.values()) + or not all(isinstance(key, str) and re.match(r"^q\d+$", key) for key in layout) + ) + + @classmethod + def _build_router(cls, router: Router | type[Router] | tuple[type[Router], dict], connectivity: nx.Graph) -> Router: + """Build a `Router` instance, given the pass router in whatever format and the connectivity graph. + + Args: + router (Router | type[Router] | tuple[type[Router], dict]): Router instance, subclass or tuple(subclass, kwargs). + connectivity (nx.Graph): Chip connectivity. + + Returns: + Router: Router instance. + + Raises: + ValueError: If the router is not a Router subclass or instance. + """ + # If router is None, we build default one: + if router is None: + return Sabre(connectivity) + + kwargs = {} + if isinstance(router, tuple): + router, kwargs = router + + # If the router is already an instance, we update the connectivity to the platform one: + if isinstance(router, Router): + if kwargs: + logger.warning("Ignoring router kwargs, as the router is already an instance.") + if isinstance(router, StarConnectivityRouter): + # For star-connectivity placers, we only care about which is the middle qubit (highest degree): + router.middle_qubit = cls._highest_degree_node(connectivity) + else: + router.connectivity = connectivity + logger.warning("Substituting the router connectivity by the transpiler/platform one.") + return router + + # If the router is a Router subclass, we instantiate it: + with contextlib.suppress(TypeError, ValueError): + if issubclass(router, Router): + if issubclass(router, StarConnectivityRouter): + # For star-connectivity placers, we only care about which is the middle qubit (highest degree): + kwargs["middle_qubit"] = cls._highest_degree_node(connectivity) + return router(connectivity, **kwargs) + + raise TypeError( + f"`router` arg ({type(router)}), must be a `Router` instance, subclass or tuple(subclass, kwargs), in `execute()`, `compile()`, `transpile_circuit()` or `route_circuit()`." + ) + + def _build_placer( + self, placer: Placer | type[Placer] | tuple[type[Placer], dict], router: Router, connectivity: nx.graph + ) -> Placer: + """Build a `Placer` instance, given the pass router in whatever format and the connectivity graph. + + Args: + router (Placer | type[Placer] | tuple[type[Placer], dict]): Placer instance, subclass or tuple(subclass, kwargs). + connectivity (nx.Graph): Chip connectivity. + + Returns: + Placer: Placer instance. + + Raises: + ValueError: If the router is not a Placer subclass or instance. + """ + # If placer is None, we build default one: + if placer is None: + return ReverseTraversal(connectivity, router) + + kwargs = {} + if isinstance(placer, tuple): + placer, kwargs = placer + + # For ReverseTraversal placer, we need to check the routing algorithm has correct connectivity: + if placer == ReverseTraversal or isinstance(placer, ReverseTraversal): + placer, kwargs = self._check_ReverseTraversal_routing_connectivity(placer, kwargs, connectivity, router) + + # If the placer is already an instance, we update the connectivity to the platform one: + if isinstance(placer, Placer): + if kwargs: + logger.warning("Ignoring placer kwargs, as the placer is already an instance.") + if isinstance(placer, StarConnectivityPlacer): + # For star-connectivity placers, we only care about which is the middle qubit (highest degree): + placer.middle_qubit = self._highest_degree_node(connectivity) + else: + placer.connectivity = connectivity + logger.warning("Substituting the placer connectivity by the transpiler/platform one.") + return placer + + # If the placer is a Placer subclass, we instantiate it: + with contextlib.suppress(TypeError, ValueError): + if issubclass(placer, Placer): + if issubclass(placer, StarConnectivityPlacer): + # For star-connectivity placers, we only care about which is the middle qubit (highest degree): + kwargs["middle_qubit"] = self._highest_degree_node(connectivity) + return placer(connectivity, **kwargs) + + raise TypeError( + f"`placer` arg ({type(placer)}), must be a `Placer` instance, subclass or tuple(subclass, kwargs), in `execute()`, `compile()`, `transpile_circuit()` or `route_circuit()`." + ) + + @staticmethod + def _check_ReverseTraversal_routing_connectivity( + placer: Placer | type[Placer], kwargs: dict, connectivity: nx.Graph, router: Router + ) -> tuple[Placer | type[Placer], dict]: + """Checks if the kwargs are valid for the Router subclass, of the ReverseTraversal Placer. + + Args: + placer (Placer | type[Placer]) + placer_kwargs (dict): kwargs for the Placer subclass. + connectivity (nx.Graph): Chip connectivity. + router (Router): Original transpilation Router subclass. + + Raises: + ValueError: If the routing_algorithm is not a Router subclass or instance. + + Returns: + tuple[Placer | type[Placer], dict]: tuple containing the final placer and its kwargs + """ + # If the placer is a ReverseTraversal instance, we update the connectivity to the platform one: + if isinstance(placer, ReverseTraversal): + placer.routing_algorithm.connectivity = connectivity + logger.warning("Substituting the ReverseTraversal router connectivity, by the transpiler/platform one.") + return placer, kwargs + + # Else is placer is not an instance, we need to check the routing algorithm: + + # If no routing algorithm is passed, we use the Transpilation Router, for the ReverseTraversal too: + if "routing_algorithm" not in kwargs: + kwargs["routing_algorithm"] = router + return placer, kwargs + + # If the routing algorithm is a Router instance, we update the connectivity to the platform one: + if isinstance(kwargs["routing_algorithm"], Router): + logger.warning( + "Substituting the passed connectivity for the ReverseTraversal routing algorithm, by the platform connectivity", + ) + kwargs["routing_algorithm"].connectivity = connectivity + return placer, kwargs + + # If the routing algorithm is a Router subclass, we instantiate it, with the platform connectivity: + with contextlib.suppress(TypeError, ValueError): + if issubclass(kwargs["routing_algorithm"], Router): + kwargs["routing_algorithm"] = kwargs["routing_algorithm"](connectivity) + return placer, kwargs + + # If the routing algorithm is not a Router subclass or instance, we raise an error: + raise TypeError( + f"`routing_algorithm` `Placer` kwarg ({kwargs['routing_algorithm']}) must be a `Router` subclass or instance, in `execute()`, `compile()`, `transpile_circuit()` or `route_circuit()`." + ) diff --git a/src/qililab/digital/circuit_to_pulses.py b/src/qililab/digital/circuit_to_pulses.py new file mode 100644 index 000000000..813cb2a48 --- /dev/null +++ b/src/qililab/digital/circuit_to_pulses.py @@ -0,0 +1,275 @@ +# Copyright 2023 Qilimanjaro Quantum Tech +# +# 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. + +"""CircuitToPulses class""" + +from dataclasses import asdict + +import numpy as np +from qibo import gates +from qibo.models import Circuit + +from qililab.constants import RUNCARD +from qililab.pulse.pulse import Pulse +from qililab.pulse.pulse_event import PulseEvent +from qililab.pulse.pulse_schedule import PulseSchedule +from qililab.settings.digital.digital_compilation_settings import DigitalCompilationSettings +from qililab.settings.digital.gate_event_settings import GateEventSettings +from qililab.typings.enums import Line +from qililab.utils import Factory + +from .native_gates import Drag, Wait + + +class CircuitToPulses: + """Translates circuits into pulse sequences.""" + + def __init__(self, digital_compilation_settings: DigitalCompilationSettings): # type: ignore # ignore typing to avoid importing platform and causing circular imports + self.digital_compilation_settings = digital_compilation_settings + + def run(self, circuit: Circuit) -> PulseSchedule: + """Translates a circuit into a pulse sequences. + + For each circuit gate we look up for its corresponding gates settings in the runcard (the name of the class of the circuit + gate and the name of the gate in the runcard should match) and load its schedule of GateEvents. + + Each gate event corresponds to a concrete pulse applied at a certain time w.r.t the gate's start time and through a specific bus + (see gates settings docstrings for more details). + + Measurement gates are handled in a slightly different manner. For a circuit gate M(0,1,2) the settings for each M(0), M(1), M(2) + will be looked up and will be applied in sync. Note that thus a circuit gate for M(0,1,2) is different from the circuit sequence + M(0)M(1)M(2) since the later will not be necessarily applied at the same time for all the qubits involved. + + Times for each qubit are kept track of with the dictionary `time`. + + The times at which each pulse is applied are padded if they are not multiples of the minimum clock time. This means that if min clock + time is 4 and a pulse applied to qubit k lasts 17ns, the next pulse at qubit k will be at t=20ns + + Args: + circuits (List[Circuit]): List of Qibo Circuit classes. + + Returns: + list[PulseSequences]: List of :class:`PulseSequences` classes. + """ + + pulse_schedule: PulseSchedule = PulseSchedule() + time: dict[int, int] = {} # init/restart time + for gate in circuit.queue: + # handle wait gates + if isinstance(gate, Wait): + self._update_time(time=time, qubit=gate.qubits[0], gate_time=gate.parameters[0]) + continue + + # Measurement gates need to be handled on their own because qibo allows to define + # an M gate as eg. gates.M(*range(5)) + if isinstance(gate, gates.M): + gate_schedule = [] + gate_qubits = gate.qubits + for qubit in gate_qubits: + gate_schedule += self._gate_schedule_from_settings(gates.M(qubit)) + + # handle control gates + else: + # extract gate schedule + gate_schedule = self._gate_schedule_from_settings(gate) + gate_qubits = self._get_gate_qubits(gate, gate_schedule) + + # process gate_schedule to pulses for both M and control gates + # get total duration for the gate + gate_time = self._get_total_schedule_duration(gate_schedule) + # update time, start time is that of the qubit most ahead in time + start_time = 0 + for qubit in gate_qubits: + start_time = max(self._update_time(time=time, qubit=qubit, gate_time=gate_time), start_time) + # sync gate end time + self._sync_qubit_times(gate_qubits, time=time) + # apply gate schedule + for gate_event in gate_schedule: + # add control gate schedule + pulse_event = self._gate_element_to_pulse_event(time=start_time, gate=gate, gate_event=gate_event) + # pop first qubit from gate if it is measurement + # this is so that the target qubit for multiM gates is every qubit in the M gate + if isinstance(gate, gates.M): + gate = gates.M(*gate.qubits[1:]) + # add event + delay = self.digital_compilation_settings.buses[gate_event.bus].delay + pulse_schedule.add_event(pulse_event=pulse_event, bus_alias=gate_event.bus, delay=delay) # type: ignore + + for bus_alias in self.digital_compilation_settings.buses: + # If we find a flux port, create empty schedule for that port. + # This is needed because for Qblox instrument working in flux buses as DC sources, if we don't + # add an empty schedule its offsets won't be activated and the results will be misleading. + if self.digital_compilation_settings.buses[bus_alias].line == Line.FLUX: + pulse_schedule.create_schedule(bus_alias=bus_alias) + + return pulse_schedule + + def _gate_schedule_from_settings(self, gate: gates.Gate) -> list[GateEventSettings]: + """Gets the gate schedule. The gate schedule is the list of pulses to apply + to a given bus for a given gate + + Args: + gate (Gate): Qibo gate + + Returns: + list[GateEventSettings]: schedule list with each of the pulses settings + """ + + gate_schedule = self.digital_compilation_settings.get_gate(name=gate.__class__.__name__, qubits=gate.qubits) + + if not isinstance(gate, Drag): + return gate_schedule + + # drag gates are currently the only parametric gates we are handling and they are handled here + if len(gate_schedule) > 1: + raise ValueError( + f"Schedule for the drag gate is expected to have only 1 pulse but instead found {len(gate_schedule)} pulses" + ) + drag_schedule = GateEventSettings( + **asdict(gate_schedule[0]) + ) # make new object so that gate_schedule is not overwritten + theta = self._normalize_angle(angle=gate.parameters[0]) + amplitude = drag_schedule.pulse.amplitude * theta / np.pi + phase = self._normalize_angle(angle=gate.parameters[1]) + if amplitude < 0: + amplitude = -amplitude + phase = self._normalize_angle(angle=gate.parameters[1] + np.pi) + drag_schedule.pulse.amplitude = amplitude + drag_schedule.pulse.phase = phase + return [drag_schedule] + + @staticmethod + def _normalize_angle(angle: float): + """Normalizes angle in range [-pi, pi]. + + Args: + angle (float): Normalized angle. + """ + angle %= 2 * np.pi + if angle > np.pi: + angle -= 2 * np.pi + return angle + + @staticmethod + def _get_total_schedule_duration(schedule: list[GateEventSettings]) -> int: + """Returns total time for a gate schedule. This is done by taking the max of (init + duration) + for all the elements in the schedule + + Args: + schedule (list[CircuitPulseSettings]): Schedule of pulses to apply + + Returns: + int: Total gate time + """ + time = 0 + for schedule_element in schedule: + time = max(time, schedule_element.pulse.duration + schedule_element.wait_time) + return time + + def _get_gate_qubits(self, gate: gates.Gate, schedule: list[GateEventSettings] | None = None) -> tuple[int, ...]: + """Gets qubits involved in gate. This includes gate.qubits but also qubits which are targets of + buses in the gate schedule + + Args: + schedule (list[CircuitPulseSettings]): Gate schedule + + Returns: + list[int]: list of qubits + """ + + schedule_qubits = ( + [ + qubit + for schedule_element in schedule + for qubit in self.digital_compilation_settings.buses[schedule_element.bus].qubits + if schedule_element.bus in self.digital_compilation_settings.buses + ] + if schedule is not None + else [] + ) + + gate_qubits = list(gate.qubits) + + return tuple(set(schedule_qubits + gate_qubits)) # convert to set and back to list to remove repeated items + + def _gate_element_to_pulse_event(self, time: int, gate: gates.Gate, gate_event: GateEventSettings) -> PulseEvent: + """Translates a gate element into a pulse. + + Args: + time (dict[int, int]): dictionary containing qubit indices as keys and current time (ns) as values + gate (gate): circuit gate. This is used only to know the qubit target of measurement gates + gate_event (GateEventSettings): gate event, a single element of a gate schedule containing information + about the pulse to be applied + bus (bus): bus through which the pulse is sent + + Returns: + PulseEvent: pulse event corresponding to the input gate event + """ + + # copy to avoid modifying runcard settings + pulse = gate_event.pulse + pulse_shape_copy = pulse.shape.copy() + pulse_shape = Factory.get(pulse_shape_copy.pop(RUNCARD.NAME))(**pulse_shape_copy) + + # handle measurement gates and target qubits for control gates which might have multi-qubit schedules + bus = self.digital_compilation_settings.buses[gate_event.bus] + qubit = ( + gate.qubits[0] + if isinstance(gate, gates.M) + else next((qubit for qubit in bus.qubits), None) + if bus is not None + else None + ) + + return PulseEvent( + pulse=Pulse( + amplitude=pulse.amplitude, + phase=pulse.phase, + duration=pulse.duration, + frequency=0, + pulse_shape=pulse_shape, + ), + start_time=time + gate_event.wait_time + self.digital_compilation_settings.delay_before_readout, + pulse_distortions=bus.distortions, + qubit=qubit, + ) + + def _update_time(self, time: dict[int, int], qubit: int, gate_time: int): + """Creates new timeline if not already created and update time. + + Args: + time (Dict[int, int]): Dictionary with the time of each qubit. + qubit_idx (int): qubit index + gate_time (int): total duration of the gate + """ + if qubit not in time: + time[qubit] = 0 + old_time = time[qubit] + residue = (gate_time) % self.digital_compilation_settings.minimum_clock_time + if residue != 0: + gate_time += self.digital_compilation_settings.minimum_clock_time - residue + time[qubit] += gate_time + return old_time + + @staticmethod + def _sync_qubit_times(qubits: list[int], time: dict[int, int]): + """Syncs the time of the given qubit list + + Args: + qubits (list[int]): qubits to sync + time (dict[int, int]): time dictionary + """ + max_time = max((time[qubit] for qubit in qubits if qubit in time), default=0) + for qubit in qubits: + time[qubit] = max_time diff --git a/src/qililab/digital/circuit_transpiler.py b/src/qililab/digital/circuit_transpiler.py index c0378684d..b3b379877 100644 --- a/src/qililab/digital/circuit_transpiler.py +++ b/src/qililab/digital/circuit_transpiler.py @@ -14,361 +14,278 @@ """Circuit Transpiler class""" -from dataclasses import asdict - -import numpy as np -from qibo import gates +import networkx as nx from qibo.models import Circuit +from qibo.transpiler.placer import Placer +from qibo.transpiler.router import Router -from qililab.constants import RUNCARD -from qililab.pulse.pulse import Pulse -from qililab.pulse.pulse_event import PulseEvent +from qililab.config import logger +from qililab.digital.circuit_optimizer import CircuitOptimizer +from qililab.digital.circuit_router import CircuitRouter +from qililab.digital.circuit_to_pulses import CircuitToPulses from qililab.pulse.pulse_schedule import PulseSchedule from qililab.settings.digital.digital_compilation_settings import DigitalCompilationSettings -from qililab.settings.digital.gate_event_settings import GateEventSettings -from qililab.typings.enums import Line -from qililab.utils import Factory from .gate_decompositions import translate_gates -from .native_gates import Drag, Wait class CircuitTranspiler: """Handles circuit transpilation. It has 3 accessible methods: - - `circuit_to_native`: transpiles a qibo circuit to native gates (Drag, CZ, Wait, M) and optionally RZ if optimize=False (optimize=True by default) - - `circuit_to_pulses`: transpiles a native gate circuit to a `PulseSchedule` - - `transpile_circuit`: runs both of the methods above sequentially + + - ``circuit_to_native``: transpiles a qibo circuit to native gates (Drag, CZ, Wait, M) and optionally RZ if optimize=False (optimize=True by default) + - ``circuit_to_pulses``: transpiles a native gate circuit to a `PulseSchedule` + - ``transpile_circuit``: runs both of the methods above sequentially + + Args: + platform (Platform): platform object containing the runcard and the chip's physical qubits. """ def __init__(self, digital_compilation_settings: DigitalCompilationSettings): # type: ignore # ignore typing to avoid importing platform and causing circular imports self.digital_compilation_settings = digital_compilation_settings - def transpile_circuit(self, circuits: list[Circuit]) -> list[PulseSchedule]: - """Transpiles a list of qibo.models.Circuit to a list of pulse schedules. - First translates the circuit to a native gate circuit and applies virtual Z gates and phase corrections for CZ gates. - Then it converts the native gate circuit to a pulse schedule using calibrated settings from the runcard. + def transpile_circuits( + self, + circuits: list[Circuit], + placer: Placer | type[Placer] | tuple[type[Placer], dict] | None = None, + router: Router | type[Router] | tuple[type[Router], dict] | None = None, + routing_iterations: int = 10, + optimize: bool = True, + ) -> tuple[list[PulseSchedule], list[dict]]: + """Transpiles a list of ``qibo.models.Circuit`` to a list of pulse schedules. - Args: - circuits (list[Circuit]): list of qibo circuits - """ - native_circuits = (self.circuit_to_native(circuit) for circuit in circuits) - return self.circuit_to_pulses(list(native_circuits)) + First makes a routing and placement of the circuit to the chip's physical qubits. And returns/logs the final layout of the qubits. - def circuit_to_native(self, circuit: Circuit, optimize: bool = True) -> Circuit: - """Converts circuit with qibo gates to circuit with native gates + Then translates the circuit to a native gate circuit and applies virtual Z gates and phase corrections for CZ gates. + + And finally, it converts the native gate circuit to a pulse schedule using calibrated settings from the runcard. + + **Examples:** + + If we instantiate some ``Circuit``, ``Platform`` and ``CircuitTranspiler`` objects like: + + .. code-block:: python + + from qibo import gates + from qibo.models import Circuit + from qibo.transpiler.placer import ReverseTraversal, Trivial + from qibo.transpiler.router import Sabre + from qililab import build_platform + from qililab.circuit_transpiler import CircuitTranspiler + + # Create circuit: + c = Circuit(5) + c.add(gates.CNOT(1, 0)) + + # Create platform: + platform = build_platform(runcard="") + + # Create transpiler: + transpiler = CircuitTranspiler(platform) + + Now we can transpile like, in the following examples: + + .. code-block:: python + + # Default Transpilation (with ReverseTraversal, Sabre, platform's connectivity and optimize = True): + routed_circuit, final_layouts = transpiler.transpile_circuits([c]) + + # Or another case, not doing optimization for some reason, and with Non-Default placer and router: + routed_circuit, final_layout = transpiler.transpile_circuits([c], placer=Trivial, router=Sabre, optimize=False) + + # Or also specifying the `router` with kwargs: + routed_circuit, final_layouts = transpiler.transpile_circuits([c], router=(Sabre, {"lookahead": 2})) Args: - circuit (Circuit): circuit with qibo gates - optimize (bool): optimize the transpiled circuit using otpimize_transpilation + circuits (list[Circuit]): list of qibo circuits. + placer (Placer | type[Placer] | tuple[type[Placer], dict], optional): `Placer` instance, or subclass `type[Placer]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `ReverseTraversal`. + router (Router | type[Router] | tuple[type[Router], dict], optional): `Router` instance, or subclass `type[Router]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `Sabre`. + routing_iterations (int, optional): Number of times to repeat the routing pipeline, to get the best stochastic result. Defaults to 10. + optimize (bool, optional): whether to optimize the circuit and/or transpilation. Defaults to True. Returns: - new_circuit (Circuit): circuit with transpiled gates + tuple[list[PulseSchedule],list[dict[str, int]]]: list of pulse schedules and list of the final layouts of the qubits, in each circuit {"qI": J}. """ - # init new circuit - new_circuit = Circuit(circuit.nqubits) - # add transpiled gates to new circuit, optimize if needed + + # Routing stage; + routed_circuits, final_layouts = zip( + *(self.route_circuit(circuit, placer, router, iterations=routing_iterations) for circuit in circuits) + ) + logger.info(f"Circuits final layouts: {final_layouts}") + + # Optimze qibo gates, cancellating redundant gates, stage: if optimize: - gates_to_optimize = translate_gates(circuit.queue) - new_circuit.add(self.optimize_transpilation(circuit.nqubits, ngates=gates_to_optimize)) - else: - new_circuit.add(translate_gates(circuit.queue)) + routed_circuits = tuple(self.optimize_circuit(circuit) for circuit in routed_circuits) - return new_circuit + # Unroll to Natives stage: + native_circuits = (self.circuit_to_native(circuit) for circuit in routed_circuits) - def optimize_transpilation(self, nqubits: int, ngates: list[gates.Gate]) -> list[gates.Gate]: - """Optimizes transpiled circuit by applying virtual Z gates. - This is done by moving all RZ to the left of all operators as a single RZ. The corresponding cumulative rotation - from each RZ is carried on as phase in all drag pulses left of the RZ operator. - Virtual Z gates are also applied to correct phase errors from CZ gates. - The final RZ operator left to be applied as the last operator in the circuit can afterwards be removed since the last - operation is going to be a measurement, which is performed on the Z basis and is therefore invariant under rotations - around the Z axis. - This last step can also be seen from the fact that an RZ operator applied on a single qubit, with no operations carried - on afterwards induces a phase rotation. Since phase is an imaginary unitary component, its absolute value will be 1 - independent on any (unitary) operations carried on it. + # Optimize native gates, optimize transpilation stage: + if optimize: + native_circuits = (self.optimize_transpilation(circuit) for circuit in native_circuits) - Mind that moving an operator to the left is equivalent to applying this operator last so - it is actually moved to the _right_ of Circuit.queue (last element of list). + # Pulse schedule stage: + pulse_schedules = self.circuit_to_pulses(list(native_circuits)) - For more information on virtual Z gates, see https://arxiv.org/abs/1612.00858 + return pulse_schedules, list(final_layouts) - Args: - nqubits (int) : number of qubits in the circuit - ngates (list[gates.Gate]) : list of gates in the circuit + def route_circuit( + self, + circuit: Circuit, + placer: Placer | type[Placer] | tuple[type[Placer], dict] | None = None, + router: Router | type[Router] | tuple[type[Router], dict] | None = None, + coupling_map: list[tuple[int, int]] | None = None, + iterations: int = 10, + ) -> tuple[Circuit, dict[str, int]]: + """Routes the virtual/logical qubits of a circuit, to the chip's physical qubits. - Returns: - list[gates.Gate] : list of re-ordered gates - """ - supported_gates = ["rz", "drag", "cz", "wait", "measure"] - new_gates = [] - shift = dict.fromkeys(range(nqubits), 0) - for gate in ngates: - if gate.name not in supported_gates: - raise NotImplementedError(f"{gate.name} not part of native supported gates {supported_gates}") - if isinstance(gate, gates.RZ): - shift[gate.qubits[0]] += gate.parameters[0] - # add CZ phase correction - elif isinstance(gate, gates.CZ): - gate_settings = self.digital_compilation_settings.get_gate( - name=gate.__class__.__name__, qubits=gate.qubits - ) - control_qubit, target_qubit = gate.qubits - corrections = next( - ( - event.pulse.options - for event in gate_settings - if ( - event.pulse.options is not None - and f"q{control_qubit}_phase_correction" in event.pulse.options - ) - ), - None, - ) - if corrections is not None: - shift[control_qubit] += corrections[f"q{control_qubit}_phase_correction"] - shift[target_qubit] += corrections[f"q{target_qubit}_phase_correction"] - new_gates.append(gate) - else: - # if gate is drag pulse, shift parameters by accumulated Zs - if isinstance(gate, Drag): - # create new drag pulse rather than modify parameters of the old one - gate = Drag(gate.qubits[0], gate.parameters[0], gate.parameters[1] - shift[gate.qubits[0]]) - - # append gate to optimized list - new_gates.append(gate) - - return new_gates - def circuit_to_pulses(self, circuits: list[Circuit]) -> list[PulseSchedule]: - """Translates a list of circuits into a list of pulse sequences (each circuit to an independent pulse sequence) - For each circuit gate we look up for its corresponding gates settings in the runcard (the name of the class of the circuit - gate and the name of the gate in the runcard should match) and load its schedule of GateEvents. - Each gate event corresponds to a concrete pulse applied at a certain time w.r.t the gate's start time and through a specific bus - (see gates settings docstrings for more details). - Measurement gates are handled in a slightly different manner. For a circuit gate M(0,1,2) the settings for each M(0), M(1), M(2) - will be looked up and will be applied in sync. Note that thus a circuit gate for M(0,1,2) is different from the circuit sequence - M(0)M(1)M(2) since the later will not be necessarily applied at the same time for all the qubits involved. + **Examples:** - Times for each qubit are kept track of with the dictionary `time`. - The times at which each pulse is applied are padded if they are not multiples of the minimum clock time. This means that if min clock - time is 4 and a pulse applied to qubit k lasts 17ns, the next pulse at qubit k will be at t=20ns + If we instantiate some ``Circuit``, ``Platform`` and ``CircuitTranspiler`` objects like: - Args: - circuits (List[Circuit]): List of Qibo Circuit classes. + .. code-block:: python - Returns: - list[PulseSequences]: List of :class:`PulseSequences` classes. - """ + from qibo import gates + from qibo.models import Circuit + from qibo.transpiler.placer import ReverseTraversal, Trivial + from qibo.transpiler.router import Sabre + from qililab import build_platform + from qililab.circuit_transpiler import CircuitTranspiler + + # Create circuit: + c = Circuit(5) + c.add(gates.CNOT(1, 0)) + + # Create platform: + platform = build_platform(runcard="") + coupling_map = platform.digital_compilation_settings.topology + + # Create transpiler: + transpiler = CircuitTranspiler(platform) + + Now we can transpile like: + + .. code-block:: python + + # Default Transpilation: + routed_circuit, final_layouts = transpiler.route_circuit([c]) # Defaults to ReverseTraversal, Sabre and platform connectivity - pulse_schedule_list: list[PulseSchedule] = [] - for circuit in circuits: - pulse_schedule = PulseSchedule() - time: dict[int, int] = {} # init/restart time - for gate in circuit.queue: - # handle wait gates - if isinstance(gate, Wait): - self._update_time(time=time, qubit=gate.qubits[0], gate_time=gate.parameters[0]) - continue - - # Measurement gates need to be handled on their own because qibo allows to define - # an M gate as eg. gates.M(*range(5)) - if isinstance(gate, gates.M): - gate_schedule = [] - gate_qubits = gate.qubits - for qubit in gate_qubits: - gate_schedule += self._gate_schedule_from_settings(gates.M(qubit)) - - # handle control gates - else: - # extract gate schedule - gate_schedule = self._gate_schedule_from_settings(gate) - gate_qubits = self._get_gate_qubits(gate, gate_schedule) - - # process gate_schedule to pulses for both M and control gates - # get total duration for the gate - gate_time = self._get_total_schedule_duration(gate_schedule) - # update time, start time is that of the qubit most ahead in time - start_time = 0 - for qubit in gate_qubits: - start_time = max(self._update_time(time=time, qubit=qubit, gate_time=gate_time), start_time) - # sync gate end time - self._sync_qubit_times(gate_qubits, time=time) - # apply gate schedule - for gate_event in gate_schedule: - # add control gate schedule - pulse_event = self._gate_element_to_pulse_event(time=start_time, gate=gate, gate_event=gate_event) - # pop first qubit from gate if it is measurement - # this is so that the target qubit for multiM gates is every qubit in the M gate - if isinstance(gate, gates.M): - gate = gates.M(*gate.qubits[1:]) - # add event - delay = self.digital_compilation_settings.buses[gate_event.bus].delay - pulse_schedule.add_event(pulse_event=pulse_event, bus_alias=gate_event.bus, delay=delay) # type: ignore - - for bus_alias in self.digital_compilation_settings.buses: - # If we find a flux port, create empty schedule for that port. - # This is needed because for Qblox instrument working in flux buses as DC sources, if we don't - # add an empty schedule its offsets won't be activated and the results will be misleading. - if self.digital_compilation_settings.buses[bus_alias].line == Line.FLUX: - pulse_schedule.create_schedule(bus_alias=bus_alias) - - pulse_schedule_list.append(pulse_schedule) - - return pulse_schedule_list - - def _gate_schedule_from_settings(self, gate: gates.Gate) -> list[GateEventSettings]: - """Gets the gate schedule. The gate schedule is the list of pulses to apply - to a given bus for a given gate + # Non-Default Trivial placer, and coupling_map specified: + routed_circuit, final_layouts = transpiler.route_circuit([c], placer=Trivial, router=Sabre, coupling_map) + + # Specifying one of the a kwargs: + routed_circuit, final_layouts = transpiler.route_circuit([c], placer=Trivial, router=(Sabre, {"lookahead": 2})) Args: - gate (Gate): Qibo gate + circuit (Circuit): circuit to route. + coupling_map (list[tuple[int, int]], optional): coupling map of the chip to route. This topology will be the one that rules, + which will overwrite any other in an instance of router or placer. Defaults to the platform topology. + placer (Placer | type[Placer] | tuple[type[Placer], dict], optional): `Placer` instance, or subclass `type[Placer]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `ReverseTraversal`. + router (Router | type[Router] | tuple[type[Router], dict], optional): `Router` instance, or subclass `type[Router]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `Sabre`. + iterations (int, optional): Number of times to repeat the routing pipeline, to keep the best stochastic result. Defaults to 10. + Returns: - list[GateEventSettings]: schedule list with each of the pulses settings + tuple[Circuit, dict[str, int]]: routed circuit, and final layout of the circuit {"qI": J}. + + Raises: + ValueError: If StarConnectivity Placer and Router are used with non-star topologies. """ + # Get the chip's connectivity + topology = nx.Graph(coupling_map if coupling_map is not None else self.digital_compilation_settings.topology) - gate_schedule = self.digital_compilation_settings.get_gate(name=gate.__class__.__name__, qubits=gate.qubits) - - if not isinstance(gate, Drag): - return gate_schedule - - # drag gates are currently the only parametric gates we are handling and they are handled here - if len(gate_schedule) > 1: - raise ValueError( - f"Schedule for the drag gate is expected to have only 1 pulse but instead found {len(gate_schedule)} pulses" - ) - drag_schedule = GateEventSettings( - **asdict(gate_schedule[0]) - ) # make new object so that gate_schedule is not overwritten - theta = self._normalize_angle(angle=gate.parameters[0]) - amplitude = drag_schedule.pulse.amplitude * theta / np.pi - phase = self._normalize_angle(angle=gate.parameters[1]) - if amplitude < 0: - amplitude = -amplitude - phase = self._normalize_angle(angle=gate.parameters[1] + np.pi) - drag_schedule.pulse.amplitude = amplitude - drag_schedule.pulse.phase = phase - return [drag_schedule] - - def _normalize_angle(self, angle: float): - """Normalizes angle in range [-pi, pi]. + circuit_router = CircuitRouter(topology, placer, router) - Args: - angle (float): Normalized angle. - """ - angle %= 2 * np.pi - if angle > np.pi: - angle -= 2 * np.pi - return angle + return circuit_router.route(circuit, iterations) + + def optimize_circuit(self, circuit: Circuit) -> Circuit: + """Main function to optimize circuits with. Currently works by cancelling adjacent hermitian gates. - def _get_total_schedule_duration(self, schedule: list[GateEventSettings]) -> int: - """Returns total time for a gate schedule. This is done by taking the max of (init + duration) - for all the elements in the schedule + The total optimization can/might be expanded in the future. Args: - schedule (list[CircuitPulseSettings]): Schedule of pulses to apply + circuit (Circuit): circuit to optimize. Returns: - int: Total gate time + Circuit: optimized circuit. """ - time = 0 - for schedule_element in schedule: - time = max(time, schedule_element.pulse.duration + schedule_element.wait_time) - return time + return CircuitOptimizer.run_gate_cancellations(circuit) - def _get_gate_qubits(self, gate: gates.Gate, schedule: list[GateEventSettings] | None = None) -> tuple[int, ...]: - """Gets qubits involved in gate. This includes gate.qubits but also qubits which are targets of - buses in the gate schedule + def circuit_to_native(self, circuit: Circuit) -> Circuit: + """Converts circuit with qibo gates to circuit with native gates Args: - schedule (list[CircuitPulseSettings]): Gate schedule + circuit (Circuit): circuit with qibo gate. Returns: - list[int]: list of qubits + new_circuit (Circuit): circuit with transpiled gates """ + new_circuit = Circuit(circuit.nqubits) + new_circuit.add(translate_gates(circuit.queue)) - schedule_qubits = ( - [ - qubit - for schedule_element in schedule - for qubit in self.digital_compilation_settings.buses[schedule_element.bus].qubits - if schedule_element.bus in self.digital_compilation_settings.buses - ] - if schedule is not None - else [] - ) + return new_circuit + + def optimize_transpilation(self, circuit: Circuit) -> Circuit: + """Optimizes transpiled circuit by applying virtual Z gates. + + This is done by moving all RZ to the left of all operators as a single RZ. The corresponding cumulative rotation + from each RZ is carried on as phase in all drag pulses left of the RZ operator. + + Virtual Z gates are also applied to correct phase errors from CZ gates. + + The final RZ operator left to be applied as the last operator in the circuit can afterwards be removed since the last + operation is going to be a measurement, which is performed on the Z basis and is therefore invariant under rotations + around the Z axis. - gate_qubits = list(gate.qubits) + This last step can also be seen from the fact that an RZ operator applied on a single qubit, with no operations carried + on afterwards induces a phase rotation. Since phase is an imaginary unitary component, its absolute value will be 1 + independent on any (unitary) operations carried on it. - return tuple(set(schedule_qubits + gate_qubits)) # convert to set and back to list to remove repeated items + Mind that moving an operator to the left is equivalent to applying this operator last so + it is actually moved to the _right_ of ``Circuit.queue`` (last element of list). - def _gate_element_to_pulse_event(self, time: int, gate: gates.Gate, gate_event: GateEventSettings) -> PulseEvent: - """Translates a gate element into a pulse. + For more information on virtual Z gates, see https://arxiv.org/abs/1612.00858 Args: - time (dict[int, int]): dictionary containing qubit indices as keys and current time (ns) as values - gate (gate): circuit gate. This is used only to know the qubit target of measurement gates - gate_event (GateEventSettings): gate event, a single element of a gate schedule containing information - about the pulse to be applied - bus (bus): bus through which the pulse is sent + circuit (Circuit): circuit with native gates, to optimize. Returns: - PulseEvent: pulse event corresponding to the input gate event + Circuit: Circuit with optimized transpiled gates. """ + optimizer = CircuitOptimizer(self.digital_compilation_settings) - # copy to avoid modifying runcard settings - pulse = gate_event.pulse - pulse_shape_copy = pulse.shape.copy() - pulse_shape = Factory.get(pulse_shape_copy.pop(RUNCARD.NAME))(**pulse_shape_copy) - - # handle measurement gates and target qubits for control gates which might have multi-qubit schedules - bus = self.digital_compilation_settings.buses[gate_event.bus] - qubit = ( - gate.qubits[0] - if isinstance(gate, gates.M) - else next((qubit for qubit in bus.qubits), None) - if bus is not None - else None - ) + output_circuit = Circuit(circuit.nqubits) + output_circuit.add(optimizer.optimize_transpilation(circuit)) + return output_circuit - return PulseEvent( - pulse=Pulse( - amplitude=pulse.amplitude, - phase=pulse.phase, - duration=pulse.duration, - frequency=0, - pulse_shape=pulse_shape, - ), - start_time=time + gate_event.wait_time + self.digital_compilation_settings.delay_before_readout, - pulse_distortions=bus.distortions, - qubit=qubit, - ) + def circuit_to_pulses(self, circuits: list[Circuit]) -> list[PulseSchedule]: + """Translates a list of circuits into a list of pulse sequences (each circuit to an independent pulse sequence). - def _update_time(self, time: dict[int, int], qubit: int, gate_time: int): - """Creates new timeline if not already created and update time. + For each circuit gate we look up for its corresponding gates settings in the runcard (the name of the class of the circuit + gate and the name of the gate in the runcard should match) and load its schedule of GateEvents. - Args: - time (Dict[int, int]): Dictionary with the time of each qubit. - qubit_idx (int): qubit index - gate_time (int): total duration of the gate - """ - if qubit not in time: - time[qubit] = 0 - old_time = time[qubit] - residue = (gate_time) % self.digital_compilation_settings.minimum_clock_time - if residue != 0: - gate_time += self.digital_compilation_settings.minimum_clock_time - residue - time[qubit] += gate_time - return old_time - - def _sync_qubit_times(self, qubits: list[int], time: dict[int, int]): - """Syncs the time of the given qubit list + Each gate event corresponds to a concrete pulse applied at a certain time w.r.t the gate's start time and through a specific bus + (see gates settings docstrings for more details). + + Measurement gates are handled in a slightly different manner. For a circuit gate M(0,1,2) the settings for each M(0), M(1), M(2) + will be looked up and will be applied in sync. Note that thus a circuit gate for M(0,1,2) is different from the circuit sequence + M(0)M(1)M(2) since the later will not be necessarily applied at the same time for all the qubits involved. + + Times for each qubit are kept track of with the dictionary `time`. + + The times at which each pulse is applied are padded if they are not multiples of the minimum clock time. This means that if min clock + time is 4 and a pulse applied to qubit k lasts 17ns, the next pulse at qubit k will be at t=20ns Args: - qubits (list[int]): qubits to sync - time (dict[int,int]): time dictionary + circuits (List[Circuit]): List of Qibo Circuit classes. + + Returns: + list[PulseSequences]: List of :class:`PulseSequences` classes. """ - max_time = max((time[qubit] for qubit in qubits if qubit in time), default=0) - for qubit in qubits: - time[qubit] = max_time + circuit_to_pulses = CircuitToPulses(self.digital_compilation_settings) + return [circuit_to_pulses.run(circuit) for circuit in circuits] diff --git a/src/qililab/execute_circuit.py b/src/qililab/execute_circuit.py index c4c602b29..cd28855b2 100644 --- a/src/qililab/execute_circuit.py +++ b/src/qililab/execute_circuit.py @@ -13,7 +13,10 @@ # limitations under the License. """Execute function used to execute a qibo Circuit using the given runcard.""" + from qibo.models import Circuit +from qibo.transpiler.placer import Placer +from qibo.transpiler.router import Router from tqdm.auto import tqdm from qililab.result import Result @@ -21,14 +24,31 @@ from .data_management import build_platform -def execute(program: Circuit | list[Circuit], runcard: str | dict, nshots: int = 1) -> Result | list[Result]: +def execute( + program: Circuit | list[Circuit], + runcard: str | dict, + nshots: int = 1, + placer: Placer | type[Placer] | tuple[type[Placer], dict] | None = None, + router: Router | type[Router] | tuple[type[Router], dict] | None = None, + routing_iterations: int = 10, + optimize: bool = True, +) -> Result | list[Result]: """Executes a Qibo circuit (or a list of circuits) with qililab and returns the results. + The ``program`` argument is first translated into pulses using the transpilation settings of the runcard and the + passed placer and router. Then the pulse will be compiled into the runcard machines assembly programs, and executed. + Args: circuit (Circuit | list[Circuit]): Qibo Circuit. runcard (str | dict): If a string, path to the YAML file containing the serialization of the Platform to be used. If a dictionary, the serialized platform to be used. nshots (int, optional): Number of shots to execute. Defaults to 1. + placer (Placer | type[Placer] | tuple[type[Placer], dict], optional): `Placer` instance, or subclass `type[Placer]` to + use`, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `ReverseTraversal`. + router (Router | type[Router] | tuple[type[Router], dict], optional): `Router` instance, or subclass `type[Router]` to + use,` with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `Sabre`. + routing_iterations (int, optional): Number of times to repeat the routing pipeline, to keep the best stochastic result. Defaults to 10. + optimize (bool, optional): whether to optimize the circuit and/or transpilation. Defaults to True. Returns: Result | list[Result]: :class:`Result` class (or list of :class:`Result` classes) containing the results of the @@ -49,12 +69,12 @@ def execute(program: Circuit | list[Circuit], runcard: str | dict, nshots: int = c = Circuit(nqubits) for qubit in range(nqubits): c.add(gates.H(qubit)) - c.add(gates.CNOT(2,0)) - c.add(gates.RY(4,np.pi / 4)) + c.add(gates.CNOT(2, 0)) + c.add(gates.RY(4, np.pi / 4)) c.add(gates.X(3)) c.add(gates.M(*range(3))) - c.add(gates.SWAP(4,2)) - c.add(gates.RX(1, 3*np.pi/2)) + c.add(gates.SWAP(4, 2)) + c.add(gates.RX(1, 3 * np.pi / 2)) probabilities = ql.execute(c, runcard="./runcards/galadriel.yml") """ @@ -68,10 +88,20 @@ def execute(program: Circuit | list[Circuit], runcard: str | dict, nshots: int = try: platform.initial_setup() platform.turn_on_instruments() - results = [] - for circuit in tqdm(program, total=len(program)): + results = [ # Execute circuit - results.append(platform.execute(circuit, num_avg=1, repetition_duration=200_000, num_bins=nshots)) + platform.execute( + circuit, + num_avg=1, + repetition_duration=200_000, + num_bins=nshots, + placer=placer, + router=router, + routing_iterations=routing_iterations, + optimize=optimize, + ) + for circuit in tqdm(program, total=len(program)) + ] platform.disconnect() return results[0] if len(results) == 1 else results except Exception as e: diff --git a/src/qililab/platform/platform.py b/src/qililab/platform/platform.py index db864c768..92f513bd3 100644 --- a/src/qililab/platform/platform.py +++ b/src/qililab/platform/platform.py @@ -69,6 +69,8 @@ if TYPE_CHECKING: from queue import Queue + from qibo.transpiler.placer import Placer + from qibo.transpiler.router import Router from qpysequence import Sequence as QpySequence from qililab.instrument_controllers.instrument_controller import InstrumentController @@ -900,11 +902,15 @@ def execute( repetition_duration: int, num_bins: int = 1, queue: Queue | None = None, + placer: Placer | type[Placer] | tuple[type[Placer], dict] | None = None, + router: Router | type[Router] | tuple[type[Router], dict] | None = None, + routing_iterations: int = 10, + optimize: bool = True, ) -> Result | QbloxResult: """Compiles and executes a circuit or a pulse schedule, using the platform instruments. If the ``program`` argument is a :class:`Circuit`, it will first be translated into a :class:`PulseSchedule` using the transpilation - settings of the platform. Then the pulse schedules will be compiled into the assembly programs and executed. + settings of the platform and the passed placer and router. Then the pulse schedules will be compiled into the assembly programs and executed. To compile to assembly programs, the ``platform.compile()`` method is called; check its documentation for more information. @@ -914,6 +920,12 @@ def execute( repetition_duration (int): Minimum duration of a single execution. num_bins (int, optional): Number of bins used. Defaults to 1. queue (Queue, optional): External queue used for asynchronous data handling. Defaults to None. + placer (Placer | type[Placer] | tuple[type[Placer], dict], optional): `Placer` instance, or subclass `type[Placer]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `ReverseTraversal`. + router (Router | type[Router] | tuple[type[Router], dict], optional): `Router` instance, or subclass `type[Router]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `Sabre`. + routing_iterations (int, optional): Number of times to repeat the routing pipeline, to keep the best stochastic result. Defaults to 10. + optimize (bool, optional): whether to optimize the circuit and/or transpilation. Defaults to True. Returns: Result: Result obtained from the execution. This corresponds to a numpy array that depending on the @@ -925,7 +937,9 @@ def execute( - Scope acquisition disabled: An array with dimension `(#sequencers, 2, #bins)`. """ # Compile pulse schedule - programs = self.compile(program, num_avg, repetition_duration, num_bins) + programs, final_layout = self.compile( + program, num_avg, repetition_duration, num_bins, placer, router, routing_iterations, optimize + ) # Upload pulse schedule for bus_alias in programs: @@ -963,12 +977,12 @@ def execute( raise ValueError("There are no readout buses in the platform.") if isinstance(program, Circuit): - results = [self._order_result(results[0], program)] + results = [self._order_result(results[0], program, final_layout)] - # FIXME: resurn result instead of results[0] + # FIXME: return result instead of results[0] return results[0] - def _order_result(self, result: Result, circuit: Circuit) -> Result: + def _order_result(self, result: Result, circuit: Circuit, final_layout: dict[str, int] | None) -> Result: """Order the results of the execution as they are ordered in the input circuit. Finds the absolute order of each measurement for each qubit and its corresponding key in the @@ -979,6 +993,7 @@ def _order_result(self, result: Result, circuit: Circuit) -> Result: Args: result (Result): Result obtained from the execution circuit (Circuit): qibo circuit being executed + final_layouts (dict[str, int]): final layout of the qubits in the circuit. Returns: Result: Result obtained from the execution, with each measurement in the same order as in circuit.queue @@ -1005,17 +1020,26 @@ def _order_result(self, result: Result, circuit: Circuit) -> Result: for qblox_result in result.qblox_raw_results: measurement = qblox_result["measurement"] qubit = qblox_result["qubit"] - results[order[qubit, measurement]] = qblox_result + original_qubit = final_layout[f"q{qubit}"] if final_layout is not None else qubit + results[order[original_qubit, measurement]] = qblox_result return QbloxResult(integration_lengths=result.integration_lengths, qblox_raw_results=results) def compile( - self, program: PulseSchedule | Circuit, num_avg: int, repetition_duration: int, num_bins: int - ) -> dict[str, list[QpySequence]]: + self, + program: PulseSchedule | Circuit, + num_avg: int, + repetition_duration: int, + num_bins: int, + placer: Placer | type[Placer] | tuple[type[Placer], dict] | None = None, + router: Router | type[Router] | tuple[type[Router], dict] | None = None, + routing_iterations: int = 10, + optimize: bool = True, + ) -> tuple[dict[str, list[QpySequence]], dict[str, int] | None]: """Compiles the circuit / pulse schedule into a set of assembly programs, to be uploaded into the awg buses. If the ``program`` argument is a :class:`Circuit`, it will first be translated into a :class:`PulseSchedule` using the transpilation - settings of the platform. Then the pulse schedules will be compiled into the assembly programs. + settings of the platform and passed placer and router. Then the pulse schedules will be compiled into the assembly programs. This methods gets called during the ``platform.execute()`` method, check its documentation for more information. @@ -1024,9 +1048,15 @@ def compile( num_avg (int): Number of hardware averages used. repetition_duration (int): Minimum duration of a single execution. num_bins (int): Number of bins used. + placer (Placer | type[Placer] | tuple[type[Placer], dict], optional): `Placer` instance, or subclass `type[Placer]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `ReverseTraversal`. + router (Router | type[Router] | tuple[type[Router], dict], optional): `Router` instance, or subclass `type[Router]` to + use, with optionally, its kwargs dict (other than connectivity), both in a tuple. Defaults to `Sabre`. + routing_iterations (int, optional): Number of times to repeat the routing pipeline, to keep the best stochastic result. Defaults to 10. + optimize (bool, optional): whether to optimize the circuit and/or transpilation. Defaults to True. Returns: - dict: Dictionary of compiled assembly programs. The key is the bus alias (``str``), and the value is the assembly compilation (``list``). + tuple[dict, dict[str, int]]: Tuple containing the dictionary of compiled assembly programs (The key is the bus alias (``str``), and the value is the assembly compilation (``list``)) and the final layout of the qubits in the circuit {"qX":Y}. Raises: ValueError: raises value error if the circuit execution time is longer than ``repetition_duration`` for some qubit. @@ -1034,15 +1064,24 @@ def compile( # We have a circular import because Platform uses CircuitToPulses and vice versa if self.digital_compilation_settings is None: raise ValueError("Cannot compile Qibo Circuit or Pulse Schedule without gates settings.") + if isinstance(program, Circuit): transpiler = CircuitTranspiler(digital_compilation_settings=self.digital_compilation_settings) - pulse_schedule = transpiler.transpile_circuit(circuits=[program])[0] + + transpiled_circuits, final_layouts = transpiler.transpile_circuits( + [program], placer, router, routing_iterations, optimize + ) + pulse_schedule, final_layout = transpiled_circuits[0], final_layouts[0] + elif isinstance(program, PulseSchedule): pulse_schedule = program + final_layout = None + else: raise ValueError( f"Program to execute can only be either a single circuit or a pulse schedule. Got program of type {type(program)} instead" ) + module_and_sequencer_per_bus: dict[str, ModuleSequencer] = { element.bus_alias: ModuleSequencer(module=instrument, sequencer=instrument.get_sequencer(channel)) for element in pulse_schedule.elements @@ -1051,10 +1090,14 @@ def compile( ) if isinstance(instrument, QbloxModule) } + compiler = PulseQbloxCompiler( buses=self.digital_compilation_settings.buses, module_and_sequencer_per_bus=module_and_sequencer_per_bus, ) - return compiler.compile( + + compiled_programs = compiler.compile( pulse_schedule=pulse_schedule, num_avg=num_avg, repetition_duration=repetition_duration, num_bins=num_bins ) + + return compiled_programs, final_layout diff --git a/tests/digital/test_circuit_optimizer.py b/tests/digital/test_circuit_optimizer.py new file mode 100644 index 000000000..c01c3a53a --- /dev/null +++ b/tests/digital/test_circuit_optimizer.py @@ -0,0 +1,129 @@ +from unittest.mock import patch +import numpy as np +from qibo import Circuit, gates + +from qililab.digital.circuit_optimizer import CircuitOptimizer +from qililab.digital.native_gates import Drag +from qililab.settings.digital.digital_compilation_settings import DigitalCompilationSettings + + +class TestCircuitOptimizerIntegration: + """Tests for the circuit optimizer class, with integration tests.""" + + def test_run_gate_cancelation(self): + """Test run gate cancelation.""" + # Create a circuit with two gates that cancel each other. + circuit = Circuit(5) + + # pairs that cancels: + circuit.add(gates.H(0)) + circuit.add(gates.H(0)) + + # From here only the X(4) will cancel with the X(4) at the end. + circuit.add(gates.CNOT(2,3)) # 1 + circuit.add(gates.X(4)) + circuit.add(gates.H(3)) # 2 + + # The 0-1 and 1-4 CNOTs shold cancel each other. + circuit.add(gates.CNOT(1,4)) + circuit.add(gates.CNOT(0,1)) + circuit.add(Drag(3, theta=2*np.pi, phase=np.pi)) # 3 + circuit.add(gates.CNOT(0,1)) + circuit.add(gates.CNOT(1,4)) + + circuit.add(gates.H(3)) # 4 + circuit.add(gates.X(4)) + circuit.add(gates.CNOT(2,3)) # 5 + + + # Optimize the circuit. + optimizer = CircuitOptimizer(None) + optimized_circuit = optimizer.run_gate_cancellations(circuit) + + # Check that the circuit is optimized + assert len(optimized_circuit.queue) == 5 + # Check name attribute: + assert [gate.name for gate in optimized_circuit.queue] == ["cx", "h", "drag", "h", "cx"] + # CHeck the type of the gates: + assert [type(gate).__name__ for gate in optimized_circuit.queue] == ["CNOT", "H", "Drag", "H", "CNOT"] + # Assert the initial arguments: + assert [gate.init_args for gate in optimized_circuit.queue] == [[2,3], [3], [3], [3], [2,3]] + assert [gate.init_kwargs for gate in optimized_circuit.queue] == [{}, {}, {"theta": 2*np.pi, "phase": np.pi, "trainable": True}, {}, {}] + + +class TestCircuitOptimizerUnit: + """Tests for the circuit optimizer class, with Unit test.""" + + @patch("qililab.digital.circuit_optimizer.CircuitOptimizer.cancel_pairs_of_hermitian_gates", return_value=[gates.CZ(0, 1), Drag(0, theta=np.pi, phase=np.pi / 2)]) + def test_run_gate_cancellations(self, mock_cancelation): + """Test optimize transpilation.""" + circuit = Circuit(2) + circuit.add(gates.RZ(0, theta=np.pi / 2)) + circuit.add(gates.CZ(0, 1)) + circuit.add(Drag(0, theta=np.pi, phase=np.pi / 2)) + + optimizer = CircuitOptimizer(None) + optimized_gates = optimizer.run_gate_cancellations(circuit) + + mock_cancelation.assert_called_once_with(circuit) + assert len(optimized_gates) == 2 + assert [gate.name for gate in optimized_gates] == ["cz", "drag"] + assert [type(gate).__name__ for gate in optimized_gates] == ["CZ", "Drag"] + + + @patch("qililab.digital.circuit_optimizer.CircuitOptimizer._create_circuit", return_value=Circuit(5)) + @patch("qililab.digital.circuit_optimizer.CircuitOptimizer._sweep_circuit_cancelling_pairs_of_hermitian_gates", return_value=[("CZ", [0, 1], {}), ("Drag", [0], {"theta": np.pi, "phase": np.pi / 2})]) + @patch("qililab.digital.circuit_optimizer.CircuitOptimizer._get_circuit_gates", return_value=[("CZ", [0, 1], {}), ("Drag", [0], {"theta": np.pi, "phase": np.pi / 2})]) + def test_cancel_pairs_of_hermitian_gates(self, mock_get_circuit_gates, mock_sweep_circuit, mock_create_circuit): + """Test run gate cancellations with mocks.""" + circuit = Circuit(2) + circuit.add(gates.RZ(0, theta=np.pi / 2)) + circuit.add(gates.CZ(0, 1)) + circuit.add(Drag(0, theta=np.pi, phase=np.pi / 2)) + + optimizer = CircuitOptimizer(None) + _ = optimizer.cancel_pairs_of_hermitian_gates(circuit) + + mock_get_circuit_gates.assert_called_once_with(circuit) + mock_sweep_circuit.assert_called_once_with([("CZ", [0, 1], {}), ("Drag", [0], {"theta": np.pi, "phase": np.pi / 2})]) + mock_create_circuit.assert_called_once_with([("CZ", [0, 1], {}), ("Drag", [0], {"theta": np.pi, "phase": np.pi / 2})], circuit.nqubits) + + + def test_get_circuit_gates(self): + """Test get circuit gates.""" + circuit = Circuit(2) + circuit.add(gates.X(0)) + circuit.add(gates.H(1)) + + gates_list = CircuitOptimizer._get_circuit_gates(circuit) + + assert gates_list == [("X", [0], {}), ("H", [1], {})] + + def test_create_gate(self): + """Test create gate.""" + gate = CircuitOptimizer._create_gate("X", [0], {}) + assert isinstance(gate, gates.X) + assert gate.init_args == [0] + + def test_create_circuit(self): + """Test create circuit.""" + gates_list = [("X", [0], {}), ("H", [1], {})] + circuit = CircuitOptimizer._create_circuit(gates_list, 2) + + assert len(circuit.queue) == 2 + assert [gate.name for gate in circuit.queue] == ["x", "h"] + + def test_sweep_circuit_cancelling_pairs_of_hermitian_gates(self): + """Test sweep circuit cancelling pairs of hermitian gates.""" + circ_list = [("X", [0], {}), ("X", [0], {}), ("H", [1], {}), ("H", [1], {})] + output_circ_list = CircuitOptimizer._sweep_circuit_cancelling_pairs_of_hermitian_gates(circ_list) + + assert output_circ_list == [] + + def test_extract_qubits(self): + """Test extract qubits.""" + qubits = CircuitOptimizer._extract_qubits([0, 1]) + assert qubits == [0, 1] + + qubits = CircuitOptimizer._extract_qubits(0) + assert qubits == [0] diff --git a/tests/digital/test_circuit_router.py b/tests/digital/test_circuit_router.py new file mode 100644 index 000000000..dd3657e51 --- /dev/null +++ b/tests/digital/test_circuit_router.py @@ -0,0 +1,348 @@ +import re +from unittest.mock import call, patch +import pytest +import networkx as nx +from qibo import Circuit, gates +from qibo.transpiler.optimizer import Preprocessing + +from qibo.transpiler.placer import ReverseTraversal, StarConnectivityPlacer, Trivial +from qibo.transpiler.router import Sabre, StarConnectivityRouter + +from qililab.digital.circuit_router import CircuitRouter + +# Different connectivities for testing the routing +linear_connectivity = [(0,1), (1,2), (2,3), (3,4)] +star_connectivity = [(0,1), (0,2), (0,3), (0,4)] + +# Different topologies for testing the routing +linear_topology = nx.Graph(linear_connectivity) +star_topology = nx.Graph(star_connectivity) + +# Linear circuit for testing specific routing +linear_circuit = Circuit(5) +linear_circuit.add(gates.H(0)) +linear_circuit.add(gates.CNOT(0, 1)) +linear_circuit.add(gates.CNOT(1, 2)) +linear_circuit.add(gates.CNOT(2, 3)) +linear_circuit.add(gates.CNOT(3, 4)) + +# Tests circuit +test_circuit = Circuit(5) +test_circuit.add(gates.H(0)) +# Tests circuit with SWAP +test_circuit_w_swap = Circuit(5) +test_circuit_w_swap.add(gates.SWAP(0,1)) +# Tests layouts +test_layout = {"q1":0} +test_bad_layout = {"q0":0, "q1":0} + +######################### +### INTEGRATION TESTS ### +######################### +class TestCircuitRouterIntegration: + """Tests for the circuit router class, integration tests.""" + + @pytest.mark.parametrize( + "type, topology", + [("linear", linear_topology), ("star", star_topology), ("bad", linear_topology)] + ) + def test_initialization(self, type, topology): + """Test the initialization of the CircuitRouter class""" + # Test the incorrect initialization of the CircuitRouter class + if type == "bad": + with pytest.raises(ValueError, match="StarConnectivity Placer and Router can only be used with star topologies"): + _ = CircuitRouter(topology, router=StarConnectivityRouter) + + # Test the correct initialization of the CircuitRouter class + elif type == "linear": + router = CircuitRouter(topology, router=Sabre, placer=ReverseTraversal) + assert isinstance(router.placer, ReverseTraversal) + assert isinstance(router.router, Sabre) + elif type == "star": + router = CircuitRouter(topology, router=StarConnectivityRouter, placer=StarConnectivityPlacer) + assert isinstance(router.placer, StarConnectivityPlacer) + assert isinstance(router.router, StarConnectivityRouter) + assert router.placer.middle_qubit == 0 + if type in ["linear", "star"]: + assert router.connectivity == topology + assert isinstance(router.preprocessing, Preprocessing) + + def test_route_doesnt_affect_already_routed_circuit(self): + """Test the routing of a circuit""" + routed_circuit, final_layout = CircuitRouter(linear_topology).route(linear_circuit) + + assert final_layout == {"q0":0, "q1":1, "q2":2, "q3":3, "q4":4} + assert routed_circuit.nqubits == linear_circuit.nqubits + assert routed_circuit.depth == linear_circuit.depth + assert [(gate.name, gate.qubits) for gate in routed_circuit.queue] == [(gate.name, gate.qubits) for gate in linear_circuit.queue] + + def test_route_affects_non_routed_circuit(self): + """Test the routing of a circuit""" + + routed_circuit, final_layout = CircuitRouter(star_topology).route(linear_circuit) + + # Assert that the circuit was routed: + assert final_layout != {"q0":0, "q1":1, "q2":2, "q3":3, "q4":4} + assert routed_circuit.nqubits == linear_circuit.nqubits + assert routed_circuit.depth > linear_circuit.depth + assert [(gate.name, gate.qubits) for gate in routed_circuit.queue] != [(gate.name, gate.qubits) for gate in linear_circuit.queue] + assert {gate.name for gate in routed_circuit.queue} >= {gate.name for gate in linear_circuit.queue} # Assert more gates + + # Assert that the circuit is routed in a concrete way: + assert final_layout == {'q0': 3, 'q1': 1, 'q2': 2, 'q3': 0, 'q4': 4} + assert routed_circuit.draw() == 'q0: ───X─o─x─X─o─\nq1: ───|─|─x─|─|─\nq2: ───|─X───o─|─\nq3: ─H─o───────|─\nq4: ───────────X─' + + +################## +### UNIT TESTS ### +################## +class TestCircuitRouterUnit: + """Tests for the circuit router class, unit tests.""" + + circuit_router = CircuitRouter(linear_topology) + + @pytest.mark.parametrize( + "type, circuit, layout, least_swaps", + [ + ("good", test_circuit, test_layout, 0), + ("none_swaos", test_circuit, test_layout, None), + ("bad_layout", test_circuit, test_bad_layout, 0) + ] + ) + @patch("qililab.config.logger.info") + @patch("qililab.digital.circuit_router.CircuitRouter._iterate_routing") + def test_route(self, mock_iterate, mock_logger_info, type, circuit, layout, least_swaps): + """Test the routing of a circuit.""" + # Set the mock returns to the parametrized test values: + mock_iterate.return_value = (circuit, layout, least_swaps) + + # Execute the routing: + if type in ["good", "none_swaos"]: + routed_circuit, final_layout = self.circuit_router.route(linear_circuit) + # Assert you return the same outputs as the mocked _iterate_routing + assert (routed_circuit, final_layout) ==(test_circuit, test_layout) + elif type == "bad_layout": + with pytest.raises(ValueError, match=re.escape(f"The final layout: {test_bad_layout} is not valid. i.e. a qubit is mapped to more than one physical qubit. Try again, if the problem persists, try another placer/routing algorithm.")): + _, _ = self.circuit_router.route(linear_circuit) + + # Assert that the logger is called properly + if type == "good": + mock_logger_info.assert_called_once_with("The best found routing, has 0 swaps.") + elif type == "none_swaos": + mock_logger_info.assert_called_once_with("No routing was done. Most probably due to routing iterations being 0.") + elif type == "bad_layout": + mock_logger_info.assert_not_called() + + # Assert that the routing pipeline was called with the correct arguments + mock_iterate.assert_called_once_with(self.circuit_router.pipeline, linear_circuit, 10) + + @pytest.mark.parametrize( + "type, circuit, layout, least_swaps, iterations", + [ + ("without_swaps", test_circuit, test_layout, None, 10), + ("with_swaps", test_circuit_w_swap, test_layout, 1, 7), + + ] + ) + @patch("qililab.digital.circuit_router.Passes.__call__") + def test_iterate_routing_without_swaps(self, mock_qibo_routing, type, circuit, layout, least_swaps, iterations): + """ Test the iterate routing of a circuit, with and without swaps.""" + # Add the mock return value to the parametrized test values: + mock_qibo_routing.return_value = (circuit, layout) + + # Execute the iterate_routing: + routed_circuit, final_layout, least_swaps = self.circuit_router._iterate_routing(self.circuit_router.pipeline, linear_circuit, iterations) + + # Assert calls on the routing algorithm: + if type == "with_swaps": + # Assert called as many times as number of iterations, since there are swaps present: + mock_qibo_routing.assert_has_calls([call(linear_circuit)]*iterations) + expected_least_swaps = 1 + elif type == "without_swaps": + # Assert only called once, since there are no swaps: + mock_qibo_routing.assert_called_once_with(linear_circuit) + expected_least_swaps = iterations = 0 # Since there are no swaps, no iterations have been needed + + # Assert you return the correct outputs: + assert (routed_circuit, final_layout, least_swaps) == (circuit, layout, expected_least_swaps) + + def test_if_star_algorithms_for_nonstar_connectivity(self): + """Test the routing of a circuit""" + circ_router = self.circuit_router + + # Assert cases where it needs to return True + assert True == circ_router._if_star_algorithms_for_nonstar_connectivity(linear_topology, StarConnectivityPlacer(), circ_router.router) + assert True == circ_router._if_star_algorithms_for_nonstar_connectivity(linear_topology, Trivial(), StarConnectivityRouter()) + assert True == circ_router._if_star_algorithms_for_nonstar_connectivity(linear_topology, StarConnectivityPlacer(), StarConnectivityRouter()) + + # Assert cases where it needs to return False + assert False == circ_router._if_star_algorithms_for_nonstar_connectivity(linear_topology, Trivial(), circ_router.router) + assert False == circ_router._if_star_algorithms_for_nonstar_connectivity(star_topology, Trivial(), StarConnectivityRouter()) + assert False == circ_router._if_star_algorithms_for_nonstar_connectivity(star_topology, circ_router.placer, circ_router.router) + + def test_highest_degree_node(self): + """Test the _highest_degree_node method.""" + # Test new edited linear topology + edited_linear_topology = nx.Graph(linear_connectivity) + edited_linear_topology.add_edges_from([(2,3),(2,4)]) + assert self.circuit_router._highest_degree_node(edited_linear_topology) == 2 + + # Test the star topology with the 0 as central + assert self.circuit_router._highest_degree_node(star_topology) == 0 + + def test_if_layout_is_not_valid(self): + """Test the _if_layout_is_not_valid method.""" + # Test valid layout + assert not self.circuit_router._if_layout_is_not_valid(test_layout) + + # Test invalid layout + assert self.circuit_router._if_layout_is_not_valid(test_bad_layout) + + @patch("qililab.digital.circuit_router.CircuitRouter._check_ReverseTraversal_routing_connectivity") + @patch("qililab.digital.circuit_router.logger.warning") + def test_build_placer(self, mock_logger_warning, mock_check_reverse): + """Test the _build_placer method.""" + + # Test default placer (ReverseTraversal) + placer = self.circuit_router._build_placer(None, self.circuit_router.router, linear_topology) + assert isinstance(placer, ReverseTraversal) + assert (placer.connectivity, placer.routing_algorithm) == (linear_topology, self.circuit_router.router) + + # Test Trivial placer + placer = self.circuit_router._build_placer(Trivial, self.circuit_router.router, linear_topology) + assert isinstance(placer, Trivial) + assert placer.connectivity == linear_topology + assert hasattr(placer, "routing_algorithm") == False + + # Test StarConnectivityPlacer with kwargs + placer = self.circuit_router._build_placer((StarConnectivityPlacer, {"middle_qubit": 0}), self.circuit_router.router, star_topology) + assert isinstance(placer, StarConnectivityPlacer) + assert placer.middle_qubit == 0 + assert hasattr(placer, "routing_algorithm") == hasattr(placer, "connectivity") == False + + # Test ReverseTraversal with kwargs + mock_check_reverse.return_value = (ReverseTraversal, {"routing_algorithm": self.circuit_router.router}) + placer = self.circuit_router._build_placer((ReverseTraversal, {"routing_algorithm": self.circuit_router.router}), self.circuit_router.router, linear_topology) + mock_check_reverse.assert_called_once_with(ReverseTraversal, {"routing_algorithm": self.circuit_router.router}, linear_topology, self.circuit_router.router) + assert isinstance(placer, ReverseTraversal) + assert (placer.connectivity, placer.routing_algorithm) == (linear_topology, self.circuit_router.router) + + # Test invalid placer type + with pytest.raises(TypeError, match="`placer` arg"): + self.circuit_router._build_placer("invalid_placer", self.circuit_router.router, linear_topology) + + # Test Placer instance, instead than subclass: + trivial_placer_instance = Trivial(linear_topology) + mock_logger_warning.reset_mock() + placer = self.circuit_router._build_placer(trivial_placer_instance, self.circuit_router.router, linear_topology) + assert isinstance(placer, Trivial) + assert placer.connectivity == linear_topology + assert hasattr(placer, "routing_algorithm") == False + mock_logger_warning.assert_has_calls([call("Substituting the placer connectivity by the transpiler/platform one.")]) + + star_placer_instance = StarConnectivityPlacer(star_topology, middle_qubit=2) + mock_logger_warning.reset_mock() + placer = self.circuit_router._build_placer(star_placer_instance, self.circuit_router.router, star_topology) + assert isinstance(placer, StarConnectivityPlacer) + assert placer.middle_qubit == 0 + assert hasattr(placer, "routing_algorithm") == hasattr(placer, "connectivity") == False + mock_logger_warning.assert_has_calls([call("Substituting the placer connectivity by the transpiler/platform one.")]) + + reverse_traversal_instance = ReverseTraversal(linear_topology, self.circuit_router.router) + placer = self.circuit_router._build_placer(reverse_traversal_instance, self.circuit_router.router, linear_topology) + assert isinstance(placer, ReverseTraversal) + assert (placer.connectivity, placer.routing_algorithm) == (linear_topology, self.circuit_router.router) + + # Test Router instance, with kwargs: + placer_instance = Trivial() + placer_kwargs = {"lookahead": 3} + mock_logger_warning.reset_mock() + router = self.circuit_router._build_placer((placer_instance,placer_kwargs),self.circuit_router.router, linear_topology) + assert hasattr(router, "lookahead") == False + mock_logger_warning.assert_has_calls([call("Ignoring placer kwargs, as the placer is already an instance."), + call("Substituting the placer connectivity by the transpiler/platform one.")]) + + @patch("qililab.digital.circuit_router.logger.warning") + def test_build_router(self, mock_logger_warning): + """Test the _build_router method.""" + + # Test default router (Sabre) + router = self.circuit_router._build_router(None, linear_topology) + assert isinstance(router, Sabre) + assert router.connectivity == linear_topology + + # Test StarConnectivityRouter + router = self.circuit_router._build_router(StarConnectivityRouter, star_topology) + assert isinstance(router, StarConnectivityRouter) + assert router.middle_qubit == 0 + assert hasattr(router, "connectivity") == False + + # Test Sabre router with kwargs + router = self.circuit_router._build_router((Sabre, {"lookahead": 2}), linear_topology) + assert isinstance(router, Sabre) + assert router.connectivity == linear_topology + assert router.lookahead == 2 + + # Test invalid router type + with pytest.raises(TypeError, match="`router` arg"): + self.circuit_router._build_router("invalid_router", linear_topology) + + # Test Router instance, instead of subclass + sabre_instance = Sabre(linear_topology) + mock_logger_warning.reset_mock() + router = self.circuit_router._build_router(sabre_instance, linear_topology) + assert isinstance(router, Sabre) + assert router.connectivity == linear_topology + mock_logger_warning.assert_has_calls([call("Substituting the router connectivity by the transpiler/platform one.")]) + + + star_router_instance = StarConnectivityRouter(star_topology) + mock_logger_warning.reset_mock() + router = self.circuit_router._build_router(star_router_instance, star_topology) + assert isinstance(router, StarConnectivityRouter) + assert router.middle_qubit == 0 + assert hasattr(router, "connectivity") == False + mock_logger_warning.assert_has_calls([call("Substituting the router connectivity by the transpiler/platform one.")]) + + # Test Router instance, with kwargs: + sabre_instance = Sabre(linear_topology, lookahead=2) + router_kwargs = {"lookahead": 3} + mock_logger_warning.reset_mock() + router = self.circuit_router._build_router((sabre_instance,router_kwargs), linear_topology) + assert router.lookahead == 2 + mock_logger_warning.assert_has_calls([call("Ignoring router kwargs, as the router is already an instance."), + call("Substituting the router connectivity by the transpiler/platform one.")]) + + def test_check_reverse_traversal_routing_connectivity(self): + """Test the _check_ReverseTraversal_routing_connectivity method.""" + # Test for a linear topology + og_placer = ReverseTraversal + og_kwargs = {"routing_algorithm": Sabre} + router = Sabre + + # Check with placer and routing algorithm both subclasses + placer, kwargs = self.circuit_router._check_ReverseTraversal_routing_connectivity(og_placer, og_kwargs, linear_topology, router) + assert (placer, kwargs) == (og_placer, og_kwargs) + + # Test for a weird router of the reversal + og_kwargs = {"routing_algorithm": int, "lookahead": 2} + with pytest.raises(TypeError, match=re.escape("`routing_algorithm` `Placer` kwarg () must be a `Router` subclass or instance, in `execute()`, `compile()`, `transpile_circuit()` or `route_circuit()`.")): + self.circuit_router._check_ReverseTraversal_routing_connectivity(og_placer, og_kwargs, star_topology, router) + + # Test that the routing_algorithm get automatically inserted: + og_kwargs = {"lookahead": 2} + placer, kwargs = self.circuit_router._check_ReverseTraversal_routing_connectivity(og_placer, og_kwargs, linear_topology, router) + assert (placer, kwargs) == (og_placer, og_kwargs | {"routing_algorithm": router}) + + # Test instance of Router, change to chips topology: + og_kwargs = {"routing_algorithm": Sabre(connectivity=None)} + placer, kwargs = self.circuit_router._check_ReverseTraversal_routing_connectivity(og_placer, og_kwargs, linear_topology, router) + og_kwargs["routing_algorithm"].connectivity = linear_topology + assert (placer, kwargs) == (og_placer, og_kwargs) + + # Test subclass of Router, change to chips topology: + og_kwargs = {"routing_algorithm": Sabre(connectivity=None)} + og_placer = ReverseTraversal(connectivity=linear_topology, routing_algorithm=og_kwargs["routing_algorithm"]) + placer, kwargs = self.circuit_router._check_ReverseTraversal_routing_connectivity(og_placer, og_kwargs, linear_topology, router) + assert og_placer.routing_algorithm.connectivity == linear_topology + assert (placer, kwargs) == (og_placer, og_kwargs) diff --git a/tests/digital/test_circuit_transpiler.py b/tests/digital/test_circuit_transpiler.py index f13b9d3a8..d9fe68ac6 100644 --- a/tests/digital/test_circuit_transpiler.py +++ b/tests/digital/test_circuit_transpiler.py @@ -1,6 +1,6 @@ import re from dataclasses import asdict -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import numpy as np import pytest @@ -9,18 +9,17 @@ from qibo.backends import NumpyBackend from qibo.gates import CZ, M, X from qibo.models import Circuit +import networkx as nx from qililab.digital import CircuitTranspiler from qililab.digital.native_gates import Drag, Wait -from qililab.platform import Bus, Buses, Platform -from qililab.pulse import Pulse, PulseEvent, PulseSchedule -from qililab.pulse.pulse_shape import SNZ, Gaussian, Rectangular -from qililab.pulse.pulse_shape import Drag as Drag_pulse -from qililab.settings import Runcard +from qililab.pulse import PulseSchedule +from qililab.settings.digital import DigitalCompilationSettings + +from qililab.digital import CircuitTranspiler +from qililab.digital.native_gates import Drag, Wait +from qililab.pulse import PulseSchedule from qililab.settings.digital import DigitalCompilationSettings -from qililab.settings.digital.gate_event_settings import GateEventSettings -from tests.data import Galadriel -from tests.test_utils import build_platform qibo.set_backend("numpy") # set backend to numpy (this is the faster option for < 15 qubits) @@ -502,7 +501,7 @@ def test_circuit_to_native(self): exhaustive=True, ) - c2 = transpiler.circuit_to_native(c1, optimize=False) + c2 = transpiler.circuit_to_native(c1) # check that both c1, c2 are qibo.Circuit assert isinstance(c1, Circuit) @@ -560,8 +559,13 @@ def test_optimize_transpilation(self, digital_settings): Drag(1, 2, -2), ] + # create circuit to test function with + circuit = Circuit(3) + circuit.add(test_gates) + # check that lists are the same - optimized_gates = transpiler.optimize_transpilation(3, test_gates) + circuit = transpiler.optimize_transpilation(circuit) + optimized_gates = list(circuit.queue) for gate_r, gate_opt in zip(result_gates, optimized_gates): assert gate_r.name == gate_opt.name assert gate_r.parameters == gate_opt.parameters @@ -640,3 +644,96 @@ def test_drag_schedule_error(self, digital_settings): transpiler = CircuitTranspiler(digital_compilation_settings=digital_settings) with pytest.raises(ValueError, match=error_string): transpiler.circuit_to_pulses(circuits=[circuit]) + + + @pytest.mark.parametrize("optimize", [True, False]) + @patch("qililab.digital.circuit_transpiler.CircuitTranspiler.optimize_circuit") + @patch("qililab.digital.circuit_transpiler.CircuitTranspiler.optimize_transpilation") + @patch("qililab.digital.circuit_transpiler.CircuitTranspiler.route_circuit") + @patch("qililab.digital.circuit_transpiler.CircuitTranspiler.circuit_to_native") + @patch("qililab.digital.circuit_transpiler.CircuitTranspiler.circuit_to_pulses") + def test_transpile_circuits(self, mock_to_pulses, mock_to_native, mock_route, mock_opt_trans, mock_opt_circuit, optimize, digital_settings): + """Test transpile_circuits method""" + transpiler = CircuitTranspiler(digital_compilation_settings=digital_settings) + placer = MagicMock() + router = MagicMock() + routing_iterations = 7 + list_size = 2 + + # Mock circuit for return values + mock_circuit = Circuit(5) + mock_circuit.add(Drag(0, 2*np.pi, np.pi)) + + # Mock layout for return values + mock_layout = {"q0": 0, "q1": 2, "q2": 1, "q3": 3, "q4": 4} + + # Mock schedule for return values + mock_schedule = PulseSchedule() + + # Mock the return values + mock_route.return_value = mock_circuit, mock_layout + mock_opt_circuit.return_value = mock_circuit + mock_to_native.return_value = mock_circuit + mock_opt_trans.return_value = mock_circuit + mock_to_pulses.return_value = [mock_schedule] + + circuit = random_circuit(5, 10, np.random.default_rng()) + + list_schedules, list_layouts = transpiler.transpile_circuits([circuit]*list_size, placer, router, routing_iterations, optimize=optimize) + + # Asserts: + # The next two functions get called for individual circuits: + mock_route.assert_called_with(circuit, placer, router, iterations=routing_iterations) + mock_to_native.assert_called_with(mock_circuit) + assert mock_route.call_count == mock_to_native.call_count == list_size + # The last one instead gets called for the whole list: + mock_to_pulses.assert_called_once_with([mock_circuit]*list_size) + assert list_schedules, list_layouts == ([mock_schedule]*list_size, [mock_layout]*list_size) + + # Asserts in optimizeL, which is called for individual circuits: + if optimize: + mock_opt_circuit.assert_called_with(mock_circuit) + mock_opt_trans.assert_called_with(mock_circuit) + assert mock_opt_circuit.call_count == mock_opt_trans.call_count == list_size + else: + mock_opt_circuit.assert_not_called() + mock_opt_trans.assert_not_called() + + @patch("qililab.digital.circuit_router.CircuitRouter.route") + def test_route_circuit(self, mock_route, digital_settings): + """Test route_circuit method""" + transpiler = CircuitTranspiler(digital_compilation_settings=digital_settings) + routing_iterations = 7 + + # Mock the return values + mock_circuit = Circuit(5) + mock_circuit.add(X(0)) + mock_layout = {"q0": 0, "q1": 2, "q2": 1, "q3": 3, "q4": 4} + mock_route.return_value = (mock_circuit, mock_layout) + + # Execute the function + circuit, layout = transpiler.route_circuit(mock_circuit, iterations=routing_iterations) + + # Asserts: + mock_route.assert_called_once_with(circuit, routing_iterations) + assert circuit, layout == (mock_circuit, mock_layout) + + @patch("qililab.digital.circuit_transpiler.nx.Graph") + @patch("qililab.digital.circuit_transpiler.CircuitRouter") + def test_that_route_circuit_instantiates_Router(self, mock_router, mock_graph, digital_settings): + """Test route_circuit method""" + transpiler = CircuitTranspiler(digital_compilation_settings=digital_settings) + routing_iterations = 7 + + # Mock the return values + mock_circuit = Circuit(5) + mock_circuit.add(X(0)) + + graph_mocking = nx.Graph(transpiler.digital_compilation_settings.topology) + mock_graph.return_value = graph_mocking + + # Execute the function + transpiler.route_circuit(mock_circuit, iterations=routing_iterations) + + # Asserts: + mock_router.assert_called_once_with(graph_mocking, None, None) diff --git a/tests/platform/test_platform.py b/tests/platform/test_platform.py index 474f90082..09e879332 100644 --- a/tests/platform/test_platform.py +++ b/tests/platform/test_platform.py @@ -528,6 +528,7 @@ def test_session_with_multiple_exceptions_in_cleanup(self): def test_compile_circuit(self, platform: Platform): """Test the compilation of a qibo Circuit.""" + circuit = Circuit(3) circuit.add(gates.X(0)) circuit.add(gates.X(1)) @@ -566,7 +567,7 @@ def test_compile_pulse_schedule(self, platform: Platform): self._compile_and_assert(platform, pulse_schedule, 2) def _compile_and_assert(self, platform: Platform, program: Circuit | PulseSchedule, len_sequences: int): - sequences = platform.compile(program=program, num_avg=1000, repetition_duration=200_000, num_bins=1) + sequences, _ = platform.compile(program=program, num_avg=1000, repetition_duration=200_000, num_bins=1) assert isinstance(sequences, dict) assert len(sequences) == len_sequences for alias, sequences_list in sequences.items(): @@ -839,24 +840,23 @@ def test_execute_returns_ordered_measurements(self, platform: Platform, qblox_re c.add([gates.M(1), gates.M(0), gates.M(0, 1)]) # without ordering, these are retrieved for each sequencer, so # the order from qblox qrm will be M(0),M(0),M(1),M(1) - platform.compile = MagicMock() # type: ignore # don't care about compilation - platform.compile.return_value = {"feedline_input_output_bus": None} - with patch.object(Bus, "upload"): - with patch.object(Bus, "run"): - with patch.object(Bus, "acquire_result") as acquire_result: - with patch.object(QbloxModule, "desync_sequencers"): - acquire_result.return_value = QbloxResult( - qblox_raw_results=qblox_results, integration_lengths=[1, 1, 1, 1] - ) - result = platform.execute(program=c, num_avg=1000, repetition_duration=2000, num_bins=1) + for idx, final_layout in enumerate([{"q0": 0, "q1": 1}, {"q0": 1, "q1": 0}]): + platform.compile = MagicMock() # type: ignore # don't care about compilation + platform.compile.return_value = {"feedline_input_output_bus": None}, final_layout + with patch.object(Bus, "upload"): + with patch.object(Bus, "run"): + with patch.object(Bus, "acquire_result") as acquire_result: + with patch.object(QbloxModule, "desync_sequencers"): + acquire_result.return_value = QbloxResult( + qblox_raw_results=qblox_results, integration_lengths=[1, 1, 1, 1] + ) + result = platform.execute(program=c, num_avg=1000, repetition_duration=2000, num_bins=1) - # check that the order of #measurement # qubit is the same as in the circuit - assert [(result["measurement"], result["qubit"]) for result in result.qblox_raw_results] == [ # type: ignore - (0, 1), - (0, 0), - (1, 0), - (1, 1), - ] + # check that the order of #measurement # qubit is the same as in the circuit + order_measurement_qubit = [(result["measurement"], result["qubit"]) for result in result.qblox_raw_results] # type: ignore + + # Change the qubit mappings, given the final_layout: + assert order_measurement_qubit == [(0, 1), (0, 0), (1, 0), (1, 1)] if idx == 0 else [(0, 0), (0, 1), (1, 1), (1, 0)] def test_execute_no_readout_raises_error(self, platform: Platform, qblox_results: list[dict]): """Test that executing with some circuit returns acquisitions with multiple measurements in same order @@ -873,7 +873,7 @@ def test_execute_no_readout_raises_error(self, platform: Platform, qblox_results # ] # in platform will be empty platform.compile = MagicMock() # type: ignore # don't care about compilation - platform.compile.return_value = {"drive_line_q0_bus": None} + platform.compile.return_value = {"drive_line_q0_bus": None}, {"q0": 0} with patch.object(Bus, "upload"): with patch.object(Bus, "run"): with patch.object(Bus, "acquire_result") as acquire_result: @@ -896,7 +896,7 @@ def test_order_results_circuit_M_neq_acquisitions(self, platform: Platform, qblo n_m = len([qubit for gate in c.queue for qubit in gate.qubits if isinstance(gate, gates.M)]) platform.compile = MagicMock() # type: ignore[method-assign] # don't care about compilation - platform.compile.return_value = {"feedline_input_output_bus": None} + platform.compile.return_value = {"feedline_input_output_bus": None}, {"q0": 0} with patch.object(Bus, "upload"): with patch.object(Bus, "run"): with patch.object(Bus, "acquire_result") as acquire_result: @@ -945,7 +945,10 @@ def test_execute_stack_2qrm(self, platform: Platform): pulse_schedule = PulseSchedule() # mock compile method platform.compile = MagicMock() # type: ignore[method-assign] - platform.compile.return_value = {"feedline_input_output_bus": None, "feedline_input_output_bus_2": None} + platform.compile.return_value = ( + {"feedline_input_output_bus": None, "feedline_input_output_bus_2": None}, + {"q0": 0}, + ) # mock execution with patch.object(Bus, "upload"): with patch.object(Bus, "run"):