"""Fidelity metrics for quantum circuit output validation.
Computes the DSR evaluation profile (success rate, chance-corrected
success, coarse TVD similarity, coarse Hellinger fidelity), the optional
Michelson "peak-contrast" DSR layer, full-histogram Hellinger fidelity/TVD,
and success rate from Sampler results, or expectation-value-based metrics
from Estimator results. Automatically detects the primitive type from the
provided inputs.
"""
from typing import Any, Callable, Dict, List, Mapping, Optional, Union
import numpy as np
from qiskit import QuantumCircuit
from qiskit_aer import AerJob
from qiskit_ibm_runtime import RuntimeJobV2
from qward.metrics.base_metric import MetricCalculator
from qward.metrics.differential_success_rate import DSRProfiler, compute_dsr_with_flags
from qward.metrics.types import MetricsId, MetricsType
from qward.schemas.fidelity_schema import FidelitySchema
JobType = Union[AerJob, RuntimeJobV2]
[docs]
class FidelityMetrics(MetricCalculator):
"""Post-runtime fidelity metrics for quantum circuit outputs.
Handles both Qiskit primitive types automatically:
- **Sampler** (counts/histograms): the DSR evaluation profile (success
rate, chance-corrected success, coarse TVD similarity, coarse
Hellinger fidelity), the optional Michelson "peak-contrast" DSR layer,
and (when a ``target_histogram`` is supplied) the full-distribution
Hellinger fidelity and TVD.
- **Estimator** (expectation values): success probability, observable
fidelity, SNR, depolarization factor
The DSR profile (``chance_corrected_success``, ``coarse_tvd_similarity``,
``coarse_hellinger_fidelity``, plus ``success_rate``) only requires
``expected_outcomes`` -- never a simulated ideal histogram. It is
computed whenever ``expected_outcomes`` is provided and
``num_measured_qubits`` can be inferred from the observed bitstrings (or
is supplied explicitly); see ``num_measured_qubits`` / ``expected_weights``
below.
Example:
# Sampler path — from counts
fm = FidelityMetrics(circuit, counts={"00": 900, "11": 100}, target_state="00")
# Sampler path — from job
fm = FidelityMetrics(circuit, job=sampler_job, expected_outcomes=["00"])
# Multi-marked target with a non-uniform reference distribution over E
fm = FidelityMetrics(
circuit,
counts=counts,
expected_outcomes=["001", "010", "100"],
expected_weights={"001": 0.5, "010": 0.25, "100": 0.25},
)
# Estimator path — from values
fm = FidelityMetrics(circuit, expectation_values=np.array([0.95]))
# Estimator path — from job (auto-detected)
fm = FidelityMetrics(circuit, job=estimator_job)
"""
def __init__(
self,
circuit: QuantumCircuit,
*,
job: Optional[Any] = None,
jobs: Optional[List[JobType]] = None,
counts: Optional[Dict[str, int]] = None,
expected_outcomes: Optional[List[str]] = None,
target_histogram: Optional[Dict[str, float]] = None,
target_state: Optional[str] = None,
success_criteria: Optional[Callable[[str], bool]] = None,
expectation_values: Optional[np.ndarray] = None,
standard_deviations: Optional[np.ndarray] = None,
ideal_expectation_values: Optional[np.ndarray] = None,
observable_labels: Optional[List[str]] = None,
num_measured_qubits: Optional[int] = None,
expected_weights: Optional[Mapping[str, float]] = None,
include_dsr_profile: bool = True,
include_michelson: bool = True,
):
if counts is not None and expectation_values is not None:
raise ValueError(
"Provide either 'counts' (Sampler) or 'expectation_values' (Estimator), not both."
)
if counts is not None and (job is not None or jobs is not None):
raise ValueError("Provide either 'job'/'jobs' or 'counts', not both.")
if expectation_values is not None and (job is not None or jobs is not None):
raise ValueError("Provide either 'job' or 'expectation_values', not both.")
self._counts = counts
self._success_criteria = success_criteria
self._jobs: List[JobType] = list(jobs) if jobs else []
if job is not None and job not in self._jobs:
self._jobs.append(job)
self._job = self._jobs[0] if self._jobs else None
self._expectation_values = expectation_values
self._standard_deviations = standard_deviations
self._ideal_expectation_values = ideal_expectation_values
self._observable_labels = observable_labels
if target_state is not None:
if expected_outcomes is None:
expected_outcomes = [target_state]
if target_histogram is None:
target_histogram = {target_state: 1.0}
self._expected_outcomes = expected_outcomes
self._target_histogram = target_histogram
self._num_measured_qubits = num_measured_qubits
self._expected_weights = expected_weights
self._include_dsr_profile = include_dsr_profile
self._include_michelson = include_michelson
super().__init__(circuit)
@property
def primitive_type(self) -> str:
"""Detected primitive type: 'sampler', 'estimator', or 'unknown'."""
if self._expectation_values is not None:
return "estimator"
if self._counts is not None:
return "sampler"
if self._job is not None:
return self._detect_job_primitive_type(self._job)
return "unknown"
def _get_metric_type(self) -> MetricsType:
return MetricsType.POST_RUNTIME
def _get_metric_id(self) -> MetricsId:
return MetricsId.FIDELITY
[docs]
def is_ready(self) -> bool:
return self.circuit is not None and (
len(self._jobs) > 0 or self._counts is not None or self._expectation_values is not None
)
[docs]
def get_metrics(self) -> Any:
"""Compute fidelity metrics based on detected primitive type.
Returns FidelitySchema for Sampler results, EstimatorSchema for Estimator results.
"""
ptype = self.primitive_type
if ptype == "estimator":
return self._compute_estimator_metrics()
return self._compute_sampler_metrics()
[docs]
def get_metrics_all(self) -> List[Any]:
"""Compute fidelity metrics for each job. Returns one schema per job.
When using counts (no jobs), returns a single-element list.
"""
if self._counts is not None or self._expectation_values is not None or len(self._jobs) <= 1:
return [self.get_metrics()]
schemas: List[Any] = []
for job in self._jobs:
ptype = self._detect_job_primitive_type(job)
if ptype == "estimator":
from qward.metrics.estimator_metrics import EstimatorMetrics
em = EstimatorMetrics(
self.circuit,
job=job,
ideal_expectation_values=self._ideal_expectation_values,
observable_labels=self._observable_labels,
)
schemas.append(em.get_metrics())
else:
counts = self._normalize_keys(self._extract_counts(job))
if not counts:
schemas.append(FidelitySchema())
continue
schemas.append(self._compute_schema(counts))
return schemas
[docs]
def add_job(self, job: Union[JobType, List[JobType]]) -> None:
"""Add one or more jobs for multi-job analysis."""
if isinstance(job, list):
for j in job:
if j not in self._jobs:
self._jobs.append(j)
elif job not in self._jobs:
self._jobs.append(job)
if not self._job and self._jobs:
self._job = self._jobs[0]
def _compute_sampler_metrics(self) -> FidelitySchema:
"""Sampler path: compute from measurement counts."""
counts = self._resolve_counts()
if not counts:
return FidelitySchema()
return self._compute_schema(counts)
def _compute_estimator_metrics(self) -> Any:
"""Estimator path: delegate to EstimatorMetrics."""
from qward.metrics.estimator_metrics import EstimatorMetrics
if self._expectation_values is not None:
em = EstimatorMetrics(
self.circuit,
expectation_values=self._expectation_values,
standard_deviations=self._standard_deviations,
ideal_expectation_values=self._ideal_expectation_values,
observable_labels=self._observable_labels,
)
else:
em = EstimatorMetrics(
self.circuit,
job=self._job,
ideal_expectation_values=self._ideal_expectation_values,
observable_labels=self._observable_labels,
)
return em.get_metrics()
@staticmethod
def _detect_job_primitive_type(job: Any) -> str:
"""Detect if a job produced Sampler or Estimator results."""
try:
result = job.result() if hasattr(job, "result") else job
if hasattr(result, "__getitem__") and len(result) > 0:
pub_result = result[0]
elif hasattr(result, "data"):
pub_result = result
else:
return "sampler"
if hasattr(pub_result, "data") and hasattr(pub_result.data, "evs"):
return "estimator"
except Exception:
pass
return "sampler"
def _compute_schema(self, counts: Dict[str, int]) -> FidelitySchema:
"""Compute fidelity schema from already-normalized counts."""
total = sum(counts.values())
unique = len(counts)
result: Dict[str, Any] = {
"shots": total,
"unique_outcomes": unique,
}
if self._expected_outcomes:
result["expected_outcomes"] = self._expected_outcomes
if self._include_michelson:
dsr, peak_mismatch = compute_dsr_with_flags(counts, self._expected_outcomes)
result["dsr"] = round(dsr, 6)
result["peak_mismatch"] = peak_mismatch
success_count = sum(counts.get(o, 0) for o in self._expected_outcomes)
result["success_rate"] = round(success_count / total, 6) if total > 0 else None
if self._include_dsr_profile:
profile_fields = self._compute_dsr_profile_fields(counts)
result.update(profile_fields)
if self._success_criteria and not self._expected_outcomes:
success_count = sum(c for state, c in counts.items() if self._success_criteria(state))
result["success_rate"] = round(success_count / total, 6) if total > 0 else None
if self._target_histogram:
observed = self._normalize_counts(counts)
hf = self._hellinger_fidelity(observed, self._target_histogram)
tvd = self._total_variation_distance(observed, self._target_histogram)
result["hellinger_fidelity"] = round(hf, 6)
result["tvd"] = round(tvd, 6)
result["tvd_fidelity"] = round(1.0 - tvd, 6)
return FidelitySchema(**result)
def _compute_dsr_profile_fields(self, counts: Dict[str, int]) -> Dict[str, Any]:
"""Compute the chance-corrected/coarse-metric components of the DSR
profile. Never raises: returns {} if the profile cannot be computed
(e.g. inconsistent bitstring lengths and no explicit
``num_measured_qubits``), since these fields are additive and must
not break existing ``FidelityMetrics`` usage.
"""
try:
profiler = DSRProfiler(
counts,
self._expected_outcomes,
num_measured_qubits=self._num_measured_qubits,
expected_weights=self._expected_weights,
include_michelson=False,
)
profile = profiler.profile()
except ValueError:
return {}
return {
"chance_baseline": profile.chance_baseline,
"chance_corrected_success": profile.chance_corrected_success,
"coarse_tvd": profile.coarse_tvd,
"coarse_tvd_similarity": profile.coarse_tvd_similarity,
"coarse_hellinger_distance": profile.coarse_hellinger_distance,
"coarse_hellinger_fidelity": profile.coarse_hellinger_fidelity,
}
def _resolve_counts(self) -> Dict[str, int]:
"""Get counts from either direct input or job extraction.
Normalizes bitstring keys by stripping spaces (Aer uses "11 00" format
when measure_all() adds a second classical register).
"""
if self._counts is not None:
raw = self._counts
elif self._job is not None:
raw = self._extract_counts(self._job)
else:
return {}
return self._normalize_keys(raw)
@staticmethod
def _normalize_keys(counts: Dict[str, int]) -> Dict[str, int]:
"""Strip spaces from bitstring keys and merge duplicates."""
normalized: Dict[str, int] = {}
for key, val in counts.items():
clean = key.replace(" ", "")
normalized[clean] = normalized.get(clean, 0) + val
return normalized
def _extract_counts(self, job: JobType) -> Dict[str, int]:
"""Extract counts from job, handling different job types."""
try:
if hasattr(job, "data") and hasattr(job.data, "get_counts"):
return job.data.get_counts()
if hasattr(job, "result"):
result = job.result()
if hasattr(result, "data") and hasattr(result.data, "get_counts"):
return result.data.get_counts()
if isinstance(job, RuntimeJobV2):
return self._extract_runtime_v2_counts(job)
result = job.result()
if hasattr(result, "get_counts"):
return result.get_counts()
return {}
except Exception:
return {}
def _extract_runtime_v2_counts(self, job: RuntimeJobV2) -> Dict[str, int]:
"""Extract counts from RuntimeJobV2."""
result = job.result()
if len(result) == 0:
return {}
pub_result = result[0]
for name in ["meas", "c", "cr", "classical"]:
if hasattr(pub_result.data, name):
register_data = getattr(pub_result.data, name)
if hasattr(register_data, "get_counts"):
return register_data.get_counts()
for attr in dir(pub_result.data):
if attr.startswith("_"):
continue
try:
register_data = getattr(pub_result.data, attr)
if hasattr(register_data, "get_counts"):
return register_data.get_counts()
except (AttributeError, TypeError):
continue
return {}
@staticmethod
def _normalize_counts(counts: Dict[str, int]) -> Dict[str, float]:
"""Convert raw counts to probability distribution."""
total = sum(counts.values())
if total == 0:
return {}
return {k: v / total for k, v in counts.items()}
@staticmethod
def _hellinger_fidelity(p: Dict[str, float], q: Dict[str, float]) -> float:
"""HF = (sum_i sqrt(p_i * q_i))^2."""
all_keys = set(p.keys()) | set(q.keys())
bc = sum(np.sqrt(p.get(k, 0.0) * q.get(k, 0.0)) for k in all_keys)
return float(min(1.0, bc**2))
@staticmethod
def _total_variation_distance(p: Dict[str, float], q: Dict[str, float]) -> float:
"""TVD = 0.5 * sum_i |p_i - q_i|."""
all_keys = set(p.keys()) | set(q.keys())
return float(0.5 * sum(abs(p.get(k, 0.0) - q.get(k, 0.0)) for k in all_keys))