diff --git a/README.md b/README.md index 77d1861..b52fc24 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ The **Quantum Simulation Environment (QSE)** is a flexible, modular, high-level Python library designed to decouple the essence of the quantum simulation problem from the technicalities of the backend software/hardware. ->> [!important] This project is under active development. +> [!important] This project is under active development. ## Architectural Overview @@ -68,7 +68,7 @@ simulation. - [myQLM](https://myqlm.github.io/), and - [Qutip](https://qutip.org/) -## 🎯 The Philosophy: Separation of Concerns +## The Philosophy: Separation of Concerns The core value of QSE is the strict separation between **Problem Framing** and **Problem Execution** @@ -77,18 +77,19 @@ The core value of QSE is the strict separation between **Problem Framing** and * | Problem Framing | `Qbits` & `Lattices` | Defining geometry, positions, and quantum degrees of freedom. | | Backend Execution | `Calculators` | Handling SDK-specific syntax, hardware constraints, and simulators. | ->> [!IMPORTANT] Why this matters ->> 1. **Backend Agnostic:** Frame your problem once; simulate it on Pulser, myQLM, or QuTiP just by switching one line of code. ->> 2. **No More "Jargon":** You don't need to learn the specific pulse sequences or gate-level syntax of every vendor to get started. You focus on the lattice and the physics. ->> 3. **Reproducibility:** Your problem definition remains a "clean" representation of the physical model, making it easier to share and verify across different research groups. +> [!IMPORTANT] Why this matters +> 1. **Backend Agnostic:** Frame your problem once; simulate it on Pulser, myQLM, or QuTiP just by switching one line of code. +> 2. **No More "Jargon":** You don't need to learn the specific pulse sequences or gate-level syntax of every vendor to get started. You focus on the lattice and the physics. +> 3. **Reproducibility:** Your problem definition remains a "clean" representation of the physical model, making it easier to share and verify across different research groups. -## 📍Position-Dependent Quantum Degrees of Freedom +## Position-Dependent Quantum Degrees of Freedom Unlike standard gate-based frameworks where qubits are abstract entities in a register, QSE treats qubits as physical objects with coordinates. This is crucial for simulations where the interaction strength between qubits is a function of their spatial separation. ### Why Positions Matter -In many physical implementations of quantum simulators—such as Rydberg Atom Arrays or Trapped Ions—the Hamiltonian of the system is governed by the distance $R_{ij}​=|{\bf R}_i - {\bf R}_j|$ between qubits $i$ and $j$: +In many physical implementations of quantum simulators—such as Rydberg Atom Arrays or Trapped Ions—the Hamiltonian of the system is governed by +the distance $R_{ij}​=|{\bf R}_i - {\bf R}_j|$ between qubits $i $ and $j $: $$ H = \sum_i \Omega_i\sigma_i^x - \sum_i \delta_i n_i + \sum_{i BaseResult: + """Common getter to safely access the attached result.""" + if self._results is None: + raise ValueError("No results available. Run calculate() first.") + return self._results + + def reset(self): + """Reset the calculator. Free up the data.""" + self._results = None + def get_spins(self): """ Get spin expectation values. @@ -139,7 +151,7 @@ def get_spins(self): if self.results is None: self.calculate() - return magnetic.get_spins(self.statevector, len(self.qbits)) + return magnetic.get_spins(self.results.statevector, len(self.qbits)) def get_sij(self): r""" @@ -158,7 +170,7 @@ def get_sij(self): if self.results is None: self.calculate() - sij = magnetic.get_sisj(self.statevector, len(self.qbits)) + sij = magnetic.get_sisj(self.results.statevector, len(self.qbits)) self.sij = sij # quick fix. TODO: proper property setup done return sij diff --git a/qse/calc/pulser.py b/qse/calc/pulser.py index 2fe0472..4c749b2 100644 --- a/qse/calc/pulser.py +++ b/qse/calc/pulser.py @@ -21,7 +21,76 @@ except ImportError: CALCULATOR_AVAILABLE = False +from functools import partial +from typing import Any, Dict, Generator +import numpy as np + +from qse.calc.results import SimResult + + +# Function to extract metadata from Pulser's simulation +def extract_safe_metadata(obj, backend_name: str) -> dict: + """ + Dynamically sweep an object for primitive attributes to build metadata. + Ignore callables, complex objects, and dunder methods. + TODO: Need to investigate some _ functions that give useful info. + """ + metadata = {"backend": backend_name} + + for attr in dir(obj): + # Optionally skip dunders, but Pulser keeps good stuff in _basis_name + if attr.startswith("__"): + continue + + try: + val = getattr(obj, attr) + # Only extract basic serializable types + if isinstance(val, (int, float, str, bool, tuple)): + # Clean up the key name (e.g., '_basis_name' -> 'basis_name') + clean_key = attr.lstrip("_") + metadata[clean_key] = val + except Exception: + pass # Ignore properties that raise errors on access + return metadata + + +# ========================================== +# GLOBAL EXTRACTORS (Pure Functions) +# ========================================== +# These functions know nothing about the Calculator. +# They only know how to translate a Pulser result into QSE format. + + +def extract_pulser_statevector(raw_results: Any) -> np.ndarray: + """Extracts the final dense complex array.""" + qobj = raw_results.get_final_state() + state_array = qobj.full().flatten() + # this is equivalent to earlier flipping based on + # the channel name == "rydberg_global" + if getattr(raw_results, "_basis_name", "") == "ground-rydberg": + state_array = state_array[::-1] + return state_array + + +def extract_pulser_counts(raw_results: Any, shots: int) -> Dict[str, int]: + """Samples the final state hardware style.""" + return dict(raw_results.sample_final_state(N_samples=shots)) + + +def extract_pulser_expectation(raw_results: Any, observable: Any) -> float: + """Delegates expectation value calculations to QuTiP.""" + exp_array = raw_results.expect([observable])[0] + return float(exp_array[-1]) + + +def extract_pulser_states(raw_results: Any) -> Generator[np.ndarray, None, None]: + """Yields the statevector at every intermediate time step.""" + for qobj in raw_results.states: + yield qobj.full().flatten() + + +# The calculator class Pulser(Calculator): r""" QSE-Calculator for pulser. @@ -118,7 +187,7 @@ def __init__( self.device = pulser.devices.MockDevice if device is None else device self.emulator = QutipEmulator if emulator is None else emulator self.wtimes = wtimes - self.results = None + self._results = None self.channel = channel self.magnetic_field = magnetic_field @@ -158,6 +227,14 @@ def register(self): def sequence(self): return self._sequence + @property + def results(self) -> SimResult: + """ + By simply overriding the property signature and calling super(), + Pylance instantly knows this branch returns a SimResult. + """ + return super().results # type: ignore + def build_sequence(self): """ Build the sequence of operations involving the qubit coordinates, @@ -182,7 +259,7 @@ def build_sequence(self): def sim(self): return self._sim - def calculate(self, progress=True): + def calculate(self, progress=True) -> SimResult: """ Run the calculation. """ @@ -191,21 +268,51 @@ def calculate(self, progress=True): if self.wtimes: t1 = time() - self.results = self.sim.run(progress_bar=progress) - - final_state = self.results.get_final_state() + # Execute the pulser backend + raw_results = self.sim.run(progress_bar=progress) + + # Extract metadata from self.sim + metadata = extract_safe_metadata(self.sim, "pulser") + # metadata = { + # "backend": "pulser", + # "basis": getattr(self.sim, "basis_name", "unknown"), + # "total_duration_ns": getattr(self.sim, "total_duration_ns", -1), + # "n_time_steps": len(self.sim.evaluation_times), + # "has_noise": hasattr(self.sim, "noise_model") + # and self.sim.noise_model is not None, + # "noise_model": getattr(self.sim, "noise_model", "unknown"), + # } + + # bind local result to the global extractor functions using "partial" + # This creates new functions/callables that already have results loaded + # as their first argument. + + bind_statevector = partial(extract_pulser_statevector, raw_results) + bind_counts = partial(extract_pulser_counts, raw_results) + bind_expectation = partial(extract_pulser_expectation, raw_results) + bind_states = partial(extract_pulser_states, raw_results) # In the qutip backend pulser uses the convention of 0 (1) being # the excited (ground) state. Hence we must reverse the state vector. - self.statevector = final_state.full().flatten() - if self.channel == "rydberg_global": - self.statevector = self.statevector[::-1] # at the moment there does not seem an effective or better alternative than - self.spins = self.get_spins() + # self.spins = self.get_spins() if self.wtimes: t2 = time() - print(f"time in compute and simulation = {t2 - t1} s.") + exec_time = t2 - t1 + metadata["exec_time"] = exec_time + print(f"time in compute and simulation = {exec_time} s.") + + # Return the clean SimResult + # SimResult can now call bound_statevector() with zero arguments! + self._results = SimResult( + statevector_func=bind_statevector, + counts_func=bind_counts, + expectation_func=bind_expectation, + states_generator=bind_states, + metadata=metadata, + ) + return self.results def _format_pulse(pulse): diff --git a/qse/calc/results.py b/qse/calc/results.py new file mode 100644 index 0000000..8781503 --- /dev/null +++ b/qse/calc/results.py @@ -0,0 +1,170 @@ +""" +Results +------- + +This module contains classes that store results of quantum simulation. + +""" + +from abc import ABC, abstractmethod +from typing import Any, Callable, Dict, Generator + +import numpy as np + + +class BaseResult(ABC): + """ + Universal class for QSE's result storage. + + The most common, core methods are listed here. + They should exist regardless of the execution path. + """ + + def __init__(self, metadata: dict = None): + self.metadata = metadata or {} + self.properties = {} # (We will use this in step 3!) + + def __repr__(self) -> str: + """Provides a clean, professional summary of the quantum execution.""" + + # 1. Identify the backend + backend = self.metadata.get("backend", "Unknown Backend").upper() + + # 2. Grab execution timing if available + exec_time = self.metadata.get("exec_time") + time_str = f"{exec_time:.4f} sec" if exec_time else "Unknown" + + # 3. Check memory states (Are the arrays loaded yet?) + # This is great for users to know if they've triggered the lazy closures + state_status = ( + "Evaluated" + if getattr(self, "_cached_state", None) is not None + else "Lazy (Pending)" + ) + + # 4. Format the output block + computed_props = list(self.properties.keys()) if self.properties else "None" + lines = [ + f"<{self.__class__.__name__} | Backend: {backend}>", + f" • Execution Time : {time_str}", + f" • Statevector : {state_status}", + f" • Metadata Keys : {len(self.metadata)} recorded", + f" • Computed Props : {computed_props}", + ] + + # Add Pulser specific flair if it exists + if "basis_name" in self.metadata: + lines.insert(2, f" • Basis : {self.metadata['basis_name']}") + + return "\n".join(lines) + + @abstractmethod + def get_counts(self, shots: int = 1024) -> Dict[str, int]: + """ + Returns bitstring frequencies. Calculated + i. via math for sim, + ii. via physical measurement for hardware + """ + pass + + @abstractmethod + def get_expectation(self, observable: Any) -> float: + """Returns expectation value. + (Exact for sim, statistical estimation for hardware) + """ + pass + + def save(self, filepath: str): + """Shared utility: Write metadata and data etc.""" + pass + + +class SimResult(BaseResult): + """ + Result from a classical emulator are stored here. It intends + to be pure, vendor-agnostic, and gives access to the quantum state. + """ + + def __init__( + self, + statevector_func: Callable, + counts_func: Callable, + expectation_func: Callable, + states_generator: Callable, + metadata: dict = None, + ): + """init function""" + super().__init__(metadata) # Hand metadata to the base class. + self._get_statevector = statevector_func + self._get_counts = counts_func + self._get_expectation = expectation_func + self._get_states = states_generator + self._statevector = None + + @property + def statevector(self) -> np.ndarray: + """Accessed as `result.statevector` + + Returns + ------- + np.ndarray + flattened array as statevector + """ + if self._statevector is None: + self._statevector = self._get_statevector() + return self._statevector + + def get_statevector(self) -> np.ndarray: + """Returns the exact dense complex array of the quantum state.""" + # Implementation: Extract and format the array from myqlm/pulser + if self._cached_state is None: + self._cached_state = self._get_statevector() + return self._cached_state + + def get_counts(self, shots: int = 1024) -> Dict[str, int]: + return self._get_counts(shots) + + def get_expectation(self, observable: Any) -> float: + """Fulfills the BaseResult abstract method contract!""" + return self._get_expectation(observable) + + def states(self) -> Generator[np.ndarray, None, None]: + """ + Yields the statevector at each time step dt. + Crucial for analyzing analog Hamiltonian evolution over time. + """ + return self._get_states() + + +class HardwareResult(BaseResult): + """ + Result from a physical QPU (e.g., Pasqal, QuEra). + Strictly limited to classical readout data and machine metadata. + """ + + def __init__(self, counts_func: Callable, metadata: Dict = None) -> None: + super().__init__(metadata) + self._get_counts = counts_func + + def get_counts(self, shots: int = 1024) -> Dict[str, int]: + # Implementation: Extract the actual + # physical measurement counts returned by the QPU + return self._get_counts(shots) + + # def get_expectation(self, observable: Any) -> float: + # Implementation: Statistical estimation based on the physical shot counts + # pass + + def get_statevector(self) -> np.ndarray: + """ + Physical QPUs cannot expose a statevector. + This method explicitly blocks unphysical requests. + """ + raise NotImplementedError( + "PhysicsError: Cannot extract an exact statevector from a physical QPU. " + "Use get_counts() to sample the physical system." + ) + + def get_machine_metadata(self) -> Dict[str, Any]: + """Returns queue time, calibration data, and hardware status.""" + pass diff --git a/uv.lock b/uv.lock index 8d23dc2..6fc99c8 100644 --- a/uv.lock +++ b/uv.lock @@ -1224,11 +1224,11 @@ wheels = [ [[package]] name = "idna" -version = "3.14" +version = "3.16" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/05/b1/efac073e0c297ecf2fb33c346989a529d4e19164f1759102dee5953ee17e/idna-3.14.tar.gz", hash = "sha256:466d810d7a2cc1022bea9b037c39728d51ae7dad40d480fc9b7d7ecf98ba8ee3", size = 198272, upload-time = "2026-05-10T20:32:15.935Z" } +sdist = { url = "https://files.pythonhosted.org/packages/1a/88/bcf9709822fe69d02c2a6a77956c98ce6ea8ca8767a9aadcedc7eb6a2390/idna-3.16.tar.gz", hash = "sha256:d7a6da03db833450fca25d2358ac9ff06cd624577a4aea3a596d5c0f77b8e03d", size = 203770, upload-time = "2026-05-22T00:16:18.781Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/6c/3c/3f62dee257eb3d6b2c1ef2a09d36d9793c7111156a73b5654d2c2305e5ce/idna-3.14-py3-none-any.whl", hash = "sha256:e677eaf072e290f7b725f9acf0b3a2bd55f9fd6f7c70abe5f0e34823d0accf69", size = 72184, upload-time = "2026-05-10T20:32:14.295Z" }, + { url = "https://files.pythonhosted.org/packages/94/16/70255075a9859a0e3adb789b68ceb0e210dec03934245fd98d248226572f/idna-3.16-py3-none-any.whl", hash = "sha256:cc246e3a3f89580c3a951b5ad298ca4638078b2cdd4f115654332b5c26daded5", size = 74165, upload-time = "2026-05-22T00:16:16.698Z" }, ] [[package]] @@ -4005,7 +4005,7 @@ wheels = [ [[package]] name = "qse" -version = "1.1.14" +version = "1.1.15" source = { editable = "." } dependencies = [ { name = "matplotlib" },