"""
Public data types for the FastQSim SDK.
All classes are pure data containers populated from FastQubit API responses.
No backend libraries (qiskit, cirq, pennylane) are imported here.
"""
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional
# ---------------------------------------------------------------------------
# Job Status
# ---------------------------------------------------------------------------
[docs]
class JobStatus(str, Enum):
"""
All possible states a FastQSim job can be in.
Terminal states: completed, failed, cancelled
Non-terminal: created, queued, running
"""
CREATED = "created"
QUEUED = "queued"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
@classmethod
def _missing_(cls, value):
raw = str(value or "").strip().lower()
if raw == "canceled":
raw = "cancelled"
for member in cls:
if member.value == raw:
return member
return None
@property
def is_terminal(self) -> bool:
"""True if the status is a final, non-changeable state."""
return self in (
JobStatus.COMPLETED,
JobStatus.FAILED,
JobStatus.CANCELLED,
)
@property
def is_successful(self) -> bool:
"""True only for COMPLETED."""
return self == JobStatus.COMPLETED
# ---------------------------------------------------------------------------
# Simulation Type
# ---------------------------------------------------------------------------
[docs]
class SimulationType(str, Enum):
"""
The simulation method used by the Q-Pod backend.
Determines both the physics model and the resource footprint:
- STATEVECTOR: Full state-vector sim. Memory ∝ 2^N. Supports all gates.
- DENSITY_MATRIX: Open-system sim with noise. Memory ∝ 4^N. Supports all gates.
- MPS: Matrix Product State. Memory ∝ N·χ². Best for low-entanglement circuits.
- CHP: Stabilizer / Clifford. Memory ∝ N². Clifford gates only (H, CNOT, S, measure).
"""
STATEVECTOR = "statevector"
DENSITY_MATRIX = "density_matrix"
MPS = "mps"
CHP = "chp"
# ---------------------------------------------------------------------------
# Internal timestamp helper
# ---------------------------------------------------------------------------
def _parse_ts(raw: Any) -> Optional[datetime]:
if raw is None:
return None
try:
return datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
except Exception:
return None
# ---------------------------------------------------------------------------
# Result
# ---------------------------------------------------------------------------
[docs]
class Result:
"""
Simulation results returned after a job reaches COMPLETED status.
Returned by Job.result() and Job.get_counts().
"""
def __init__(self, data: Dict[str, Any]) -> None:
self._data = data
@property
def counts(self) -> Dict[str, int]:
"""Measurement counts dictionary, e.g. {'00': 512, '11': 512}."""
return self._data.get("counts", {})
@property
def probabilities(self) -> Dict[str, float]:
"""Outcome probabilities, e.g. {'00': 0.5, '11': 0.5}."""
return self._data.get("probabilities", {})
@property
def statevector(self) -> Optional[List[complex]]:
"""
Final statevector as a list of complex amplitudes, if available.
None when statevector was not requested or is too large.
"""
return self._data.get("statevector")
@property
def memory(self) -> Optional[List[str]]:
"""
Individual shot results as bit-strings, e.g. ['00', '11', '00', ...].
Only populated when the job was submitted with memory=True.
"""
return self._data.get("memory")
@property
def execution_time(self) -> float:
"""Wall-clock execution time in seconds as measured by the Q-Pod."""
return self._data.get("execution_time", 0.0)
@property
def metadata(self) -> Dict[str, Any]:
"""Job and result metadata returned by the server."""
return self._data.get("metadata", {})
[docs]
def get_counts(self, key: Optional[str] = None) -> Dict[str, int]:
"""
Get counts for a specific measurement register, or all counts.
Args:
key: Register name. If None, returns the full counts dict.
"""
if key is None:
return self.counts
return self._data.get("counts_by_register", {}).get(key, {})
def __repr__(self) -> str:
return f"Result(counts={self.counts}, execution_time={self.execution_time}s)"
# ---------------------------------------------------------------------------
# Job
# ---------------------------------------------------------------------------
[docs]
class Job:
"""
Represents a quantum simulation job submitted to the FastQSim platform.
Returned by FastQSimClient.run() and FastQSimClient.run_batch().
Call result() to block until completion and retrieve results, or use
wait_for_completion() to poll and then inspect properties manually.
"""
def __init__(self, data: Dict[str, Any], client: Any = None) -> None:
"""
Args:
data: Raw JSON response dict from the FastQubit API (JobResponse).
client: FastQSimClient instance, used for polling methods.
"""
self._data = data
self._client = client
# ------------------------------------------------------------------
# Identity
# ------------------------------------------------------------------
@property
def job_id(self) -> str:
"""Unique job identifier (UUID)."""
return self._data.get("job_id", "")
@property
def end_user_id(self) -> str:
"""End-user identity asserted by the integrator who owns this job."""
return self._data.get("end_user_id", "")
@property
def integrator_id(self) -> Optional[str]:
"""Integrator that submitted this job."""
return self._data.get("integrator_id")
@property
def session_id(self) -> str:
"""Session under which this job was submitted."""
return self._data.get("session_id", "")
@property
def job_name(self) -> Optional[str]:
"""Optional human-readable job label supplied at submission time."""
return self._data.get("job_name")
@property
def tags(self) -> Dict[str, Any]:
"""Key-value tags attached to the job at submission time."""
return self._data.get("tags") or {}
@property
def metadata(self) -> Dict[str, Any]:
"""Opaque client metadata stored with the job at submission time."""
return self._data.get("metadata") or {}
# ------------------------------------------------------------------
# Status
# ------------------------------------------------------------------
@property
def status(self) -> JobStatus:
"""Current job status as a JobStatus enum value."""
return JobStatus(self._data.get("status", "queued"))
@property
def ok(self) -> bool:
"""True if the job's current status is COMPLETED."""
return self.status == JobStatus.COMPLETED
@property
def error_message(self) -> Optional[str]:
"""
Worker error message if the job failed.
Contains the raw exception text from the Q-Pod (e.g. ParseError details).
"""
return self._data.get("error_message")
# ------------------------------------------------------------------
# Execution config
# ------------------------------------------------------------------
@property
def backend(self) -> str:
"""Backend used for execution ('qiskit', 'cirq', 'pennylane')."""
return self._data.get("backend", "")
@property
def device(self) -> str:
"""Device used for execution ('cpu', 'gpu', 'quantum')."""
return self._data.get("device", "cpu")
@property
def simulation_type(self) -> str:
"""Simulation mode used by the backend ('statevector', 'density_matrix', 'mps', 'chp')."""
return self._data.get("simulation_type", "statevector")
@property
def shots(self) -> Optional[int]:
"""Number of measurement shots specified at submission time."""
return self._data.get("shots")
@property
def seed(self) -> Optional[int]:
"""Optional deterministic RNG seed supplied at submission time."""
return self._data.get("seed")
@property
def options(self) -> Dict[str, Any]:
"""Backend/device-specific execution options supplied at submission time."""
return self._data.get("options") or {}
# ------------------------------------------------------------------
# Circuit profile (populated once AST analyzer is wired in)
# ------------------------------------------------------------------
@property
def profile(self) -> Dict[str, Any]:
"""
Circuit complexity metrics extracted by the AST analyzer.
Fields: qubit_count, depth, gate_count, two_qubit_gate_count.
All None until the analyzer is implemented.
"""
return self._data.get("profile") or {}
@property
def num_qubits(self) -> Optional[int]:
"""Number of qubits in the circuit (from circuit profile). None until analyzer is wired."""
return (self._data.get("profile") or {}).get("qubit_count")
# ------------------------------------------------------------------
# Resource estimate
# ------------------------------------------------------------------
@property
def resource_estimate(self) -> Dict[str, Any]:
"""
Predicted Kubernetes resource allocation for the worker pod.
Fields: cpu_request, memory_request, cpu_limit, memory_limit.
All None until adaptive sizing is implemented.
"""
return self._data.get("resource_estimate") or {}
# ------------------------------------------------------------------
# Result data
# ------------------------------------------------------------------
@property
def result_counts(self) -> Optional[Dict[str, int]]:
"""Inline measurement counts (bit-string → count). Populated when job completes."""
return self._data.get("result_counts")
@property
def result_metadata(self) -> Dict[str, Any]:
"""
Additional result metadata emitted by the worker (backend name,
simulation_type, engine version, etc.).
"""
return self._data.get("result_metadata") or {}
@property
def has_external_artifact(self) -> bool:
"""
True when the result was too large for inline storage and is persisted to S3.
Use GET /jobs/{job_id}/result-url (client._fetch_external_artifact) to retrieve it.
"""
return bool(self._data.get("has_external_artifact", False))
@property
def result_url_expires_at(self) -> Optional[datetime]:
"""Expiry of the last issued presigned result download URL."""
return _parse_ts(self._data.get("result_url_expires_at"))
# ------------------------------------------------------------------
# Timing
# ------------------------------------------------------------------
@property
def created_at(self) -> Optional[datetime]:
"""UTC time when the job record was first created."""
return _parse_ts(self._data.get("created_at"))
@property
def updated_at(self) -> Optional[datetime]:
"""UTC time when the job record was last modified."""
return _parse_ts(self._data.get("updated_at"))
@property
def queue_enqueued_at(self) -> Optional[datetime]:
"""When the job was accepted and published to the RabbitMQ queue."""
return _parse_ts(self._data.get("queue_enqueued_at"))
@property
def queue_dequeued_at(self) -> Optional[datetime]:
"""When the scheduler claimed the job and a worker pod was created."""
return _parse_ts(self._data.get("queue_dequeued_at"))
@property
def execution_started_at(self) -> Optional[datetime]:
"""Timestamp emitted by the q-worker when simulation begins inside the pod."""
return _parse_ts(self._data.get("execution_started_at"))
@property
def execution_finished_at(self) -> Optional[datetime]:
"""Timestamp emitted by the q-worker after simulation and artifact upload."""
return _parse_ts(self._data.get("execution_finished_at"))
@property
def completed_at(self) -> Optional[datetime]:
"""When the worker finalized the terminal state in persistent storage."""
return _parse_ts(self._data.get("completed_at"))
# ------------------------------------------------------------------
# Execution metrics
# ------------------------------------------------------------------
@property
def execution_time_seconds(self) -> Optional[float]:
"""Worker-reported wall-clock execution duration in seconds."""
return self._data.get("execution_time_seconds")
@property
def cpu_seconds_total(self) -> Optional[float]:
"""Worker-reported process CPU time (user + system) consumed during execution."""
return self._data.get("cpu_seconds_total")
@property
def peak_memory_mb(self) -> Optional[float]:
"""Worker-reported peak RSS memory usage in MB sampled during execution."""
return self._data.get("peak_memory_mb")
# ------------------------------------------------------------------
# Billing telemetry
# ------------------------------------------------------------------
@property
def billing_avg_memory_mb(self) -> Optional[float]:
"""Billing telemetry: average RSS memory usage in MB over execution."""
return self._data.get("billing_avg_memory_mb")
@property
def billing_cpu_millicore_seconds(self) -> Optional[float]:
"""Billing telemetry: integrated CPU usage in millicore-seconds."""
return self._data.get("billing_cpu_millicore_seconds")
@property
def billing_memory_gb_seconds(self) -> Optional[float]:
"""Billing telemetry: integrated memory usage in GB-seconds."""
return self._data.get("billing_memory_gb_seconds")
@property
def billing_psutil_available(self) -> Optional[bool]:
"""Whether psutil-based billing telemetry was available in the worker runtime."""
return self._data.get("billing_psutil_available")
@property
def billing_memory_headroom_mb(self) -> Optional[float]:
"""Billing telemetry: memory headroom to pod limit in MB at completion."""
return self._data.get("billing_memory_headroom_mb")
# ------------------------------------------------------------------
# Infrastructure
# ------------------------------------------------------------------
@property
def pod_name(self) -> Optional[str]:
"""Kubernetes pod name used for this job execution."""
return self._data.get("pod_name")
# ------------------------------------------------------------------
# Methods (polling and result retrieval delegate to client)
# ------------------------------------------------------------------
[docs]
def result(self, timeout: Optional[int] = None) -> "Result":
"""
Block until the job completes, then return the Result.
Args:
timeout: Maximum seconds to wait. Raises JobTimeoutError if exceeded.
Raises:
JobTimeoutError: Job did not complete within timeout.
JobFailedError: Job reached a failed terminal state.
"""
if self._client is None:
raise RuntimeError("Job has no client attached; cannot poll for result.")
return self._client._wait_and_get_result(self.job_id, timeout=timeout)
[docs]
def wait_for_completion(
self,
poll_interval: int = 5,
timeout: Optional[int] = None,
) -> None:
"""
Block until the job reaches a terminal state.
Args:
poll_interval: Seconds between status checks.
timeout: Maximum seconds to wait. Raises JobTimeoutError if exceeded.
"""
if self._client is None:
raise RuntimeError("Job has no client attached; cannot poll.")
self._client._poll_until_terminal(
self.job_id, poll_interval=poll_interval, timeout=timeout
)
[docs]
def get_counts(self) -> Optional[Dict[str, int]]:
"""
Return measurement counts for a completed job.
Raises:
JobNotCompleteError: Job has not finished yet.
JobDoesNotContainCountsError: Job was run without measurement shots.
"""
if self._client is None:
raise RuntimeError("Job has no client attached.")
return self._client._get_counts(self.job_id)
[docs]
def get_statevector(self) -> List[complex]:
"""
Retrieve the statevector for a completed job.
Raises:
JobNotCompleteError: Job has not finished yet.
JobStatevectorTooLargeError: Statevector exceeds retrieval limit.
"""
if self._client is None:
raise RuntimeError("Job has no client attached.")
return self._client._get_statevector(self.job_id)
[docs]
def cancel(self) -> bool:
"""
Request cancellation of a QUEUED or RUNNING job.
Returns:
True if the cancellation was accepted.
Raises:
JobCouldNotCancelError: Server rejected the cancellation request.
"""
if self._client is None:
raise RuntimeError("Job has no client attached.")
return self._client._cancel_job(self.job_id)
[docs]
def refresh(self) -> "Job":
"""Fetch the latest metadata for this job and update in place."""
if self._client is None:
raise RuntimeError("Job has no client attached.")
updated = self._client.get(self.job_id)
self._data = updated._data
return self
def __repr__(self) -> str:
return f"Job(job_id='{self.job_id}', status={self.status.value!r})"
# ---------------------------------------------------------------------------
# Session
# ---------------------------------------------------------------------------
[docs]
class Session:
"""
Represents a FastQubit session payload.
In pod-token SDK mode, session lifecycle operations are handled by the
integrator control plane, so helper methods like refresh_token() and
close() may raise ValidationError.
"""
def __init__(self, data: Dict[str, Any], client: Any = None) -> None:
self._data = data
self._client = client
@property
def session_id(self) -> str:
"""Unique session identifier (UUID)."""
return self._data.get("session_id", "")
@property
def end_user_id(self) -> str:
"""End-user identity asserted by the integrator who owns this session."""
return self._data.get("end_user_id", "")
@property
def integrator_id(self) -> Optional[str]:
"""Integrator that created this session."""
return self._data.get("integrator_id")
@property
def state(self) -> str:
"""
Current session lifecycle state.
One of: creating | provisioning | ready | active | idle | terminating | terminated | failed.
"""
return self._data.get("state", "active")
@property
def execution_engine(self) -> str:
"""Runtime engine used inside the user pod (e.g. 'jupyter-kernel-gateway')."""
return self._data.get("execution_engine", "jupyter-kernel-gateway")
@property
def kernel_id(self) -> Optional[str]:
"""Jupyter kernel ID. Populated after the pod is ready. None for terminal sessions."""
return self._data.get("kernel_id")
@property
def ws_url(self) -> Optional[str]:
"""Fully resolved WSS URL for the active kernel channel. None until pod is ready."""
return self._data.get("ws_url")
@property
def ws_token(self) -> Optional[str]:
"""Short-lived HMAC-signed session token for the WebSocket connection."""
return self._data.get("ws_token")
@property
def ws_token_expires_at(self) -> Optional[datetime]:
"""UTC expiry of the current ws_token."""
return _parse_ts(self._data.get("ws_token_expires_at"))
@property
def expires_at(self) -> Optional[datetime]:
"""Hard TTL for this session. The pod is terminated after this time."""
return _parse_ts(self._data.get("expires_at"))
@property
def created_at(self) -> Optional[datetime]:
"""UTC time when the session was created."""
return _parse_ts(self._data.get("created_at"))
@property
def error_message(self) -> Optional[str]:
"""Error detail when state is 'failed'."""
return self._data.get("error_message")
@property
def already_active(self) -> Optional[bool]:
"""
Present on POST /sessions/start responses only.
True when an existing active session was returned because the per-user
limit was already reached. False when a new session was provisioned.
"""
return self._data.get("already_active")
@property
def message(self) -> Optional[str]:
"""Human-readable note, e.g. instructions to terminate before starting a new session."""
return self._data.get("message")
[docs]
def refresh_token(self) -> None:
"""Request a refreshed WebSocket authentication token from the server.
Raises ValidationError in pod-token SDK mode.
"""
if self._client is None:
raise RuntimeError("Session has no client attached.")
self._client._refresh_token(self.session_id)
[docs]
def close(self) -> None:
"""Terminate the session and release server-side resources.
Raises ValidationError in pod-token SDK mode.
"""
if self._client is None:
return
self._client._close_session(self.session_id)
def __repr__(self) -> str:
return (
f"Session(session_id='{self.session_id}', "
f"end_user_id='{self.end_user_id}', "
f"state='{self.state}')"
)
# ---------------------------------------------------------------------------
# RunArgs (convenience helper for programmatic job construction)
# ---------------------------------------------------------------------------
[docs]
class RunArgs:
"""
Optional helper to construct job submission arguments.
Can be passed to FastQSimClient.run() instead of keyword arguments.
All fields mirror the run() parameters exactly.
"""
def __init__(
self,
circuit: str,
backend: str = "qiskit",
device: str = "cpu",
shots: int = 1024,
simulation_type: Optional[str] = None,
seed: Optional[int] = None,
metadata: Optional[Dict[str, Any]] = None,
job_name: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
tags: Optional[Dict[str, Any]] = None,
) -> None:
self.circuit = circuit
self.backend = backend
self.device = device
self.shots = shots
self.simulation_type = simulation_type
self.seed = seed
self.metadata = metadata or {}
self.job_name = job_name
self.options = options or {}
self.tags = tags or {}
[docs]
def to_dict(self) -> Dict[str, Any]:
"""Serialize to a dictionary for passing to client.run(**args.to_dict())."""
d: Dict[str, Any] = {
"circuit": self.circuit,
"backend": self.backend,
"device": self.device,
"shots": self.shots,
"seed": self.seed,
"metadata": self.metadata,
"job_name": self.job_name,
"options": self.options,
"tags": self.tags,
}
if self.simulation_type is not None:
d["simulation_type"] = self.simulation_type
return d
def __repr__(self) -> str:
return (
f"RunArgs(backend='{self.backend}', device='{self.device}', "
f"shots={self.shots}, simulation_type={self.simulation_type!r})"
)