Skip to content

Python SDK

Classiq SDK.

analyzer special

analyzer

Analyzer module, implementing facilities for analyzing circuits using Classiq platform.

Analyzer

Analyzer is the wrapper object for all analysis capabilities.

Source code in classiq/analyzer/analyzer.py
class Analyzer(metaclass=Asyncify):
    """Analyzer is the wrapper object for all analysis capabilities."""

    def __init__(self, circuit: generator_result.GeneratedCircuit):
        """Init self.

        Args:
            circuit (): The circuit to be analyzed.
        """
        self.graph: go.Figure = None
        self.hardware_comparison_table: go.Figure = None
        if circuit.qasm is None:
            raise ValueError("Analysis requires a circuit with valid QASM code")
        self._params = analysis_params.AnalysisParams(qasm=circuit.qasm)
        self.input = circuit

    async def analyze_async(self) -> analysis_result.Analysis:
        """Runs the circuit analysis.

        Returns:
            The analysis result.
        """
        result = await ApiWrapper.call_analysis_task(self._params)

        if result.status != analysis_result.AnalysisStatus.SUCCESS:
            raise ClassiqAnalyzerError(f"Analysis failed: {result.details}")
        details = validate_type(
            obj=result.details,
            expected_type=analysis_result.Analysis,
            operation="Analysis",
            exception_type=ClassiqAnalyzerError,
        )

        dashboard_path = routes.ANALYZER_DASHBOARD
        self._open_route(path=dashboard_path)
        return details

    async def analyzer_app_async(self) -> None:
        """Opens the analyzer app with synthesis interactive results.

        Returns:
            None.
        """
        result = await ApiWrapper.call_analyzer_app(self.input)
        webbrowser.open_new_tab(urljoin(routes.ANALYZER_FULL_FE_APP, str(result.id)))

    async def get_qubits_connectivity_async(self) -> None:
        """create a network connectivity graph of the analysed circuit.

        Returns:
            None.
        """
        result = await ApiWrapper.call_graphs_task(self._params)
        self.graph = go.Figure(json.loads(result.graph))

    async def plot_qubits_connectivity_async(self) -> None:
        """plot the connectivity graph. if it has not been created it, it first creates the graph.

        Returns:
            None.
        """
        if not hasattr(self, "graph"):
            await self.get_qubits_connectivity_async()
        self.graph.show()

    async def get_hardware_comparison_table_async(
        self,
        hardware: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
    ) -> None:
        """create a comparison table between the transpiled circuits result on different hardware.
        The  comparison table included the depth, multi qubit gates count,and total gates count of the circuits.

        Args: hardware (): List of hardware (string or `AnalyzerProviderVendor`). if None, the table will include all
        the available hardware.

        Returns: None.
        """
        if hardware is None:
            hardware = list(AnalyzerProviderVendor)
        params = analysis_params.AnalysisTableParams(
            qasm=self._params.qasm, hardware=hardware
        )
        result = await ApiWrapper.call_table_graphs_task(params=params)
        self.hardware_comparison_table = go.Figure(json.loads(result.graph))

    async def plot_hardware_comparison_table_async(
        self,
    ) -> None:
        """plot the comparison table. if it has not been created it, it first creates the table using all the
        available hardware.

        Returns:
            None.
        """
        if not hasattr(self, "hardware_comparison_table"):
            await self.get_hardware_comparison_table_async()
        self.hardware_comparison_table.show()

    @staticmethod
    def _open_route(path: str) -> None:
        backend_uri = client.client().get_backend_uri()
        webbrowser.open_new_tab(f"{backend_uri}{path}")
__init__(self, circuit) special

Init self.

Parameters:

Name Type Description Default
circuit

The circuit to be analyzed.

required
Source code in classiq/analyzer/analyzer.py
def __init__(self, circuit: generator_result.GeneratedCircuit):
    """Init self.

    Args:
        circuit (): The circuit to be analyzed.
    """
    self.graph: go.Figure = None
    self.hardware_comparison_table: go.Figure = None
    if circuit.qasm is None:
        raise ValueError("Analysis requires a circuit with valid QASM code")
    self._params = analysis_params.AnalysisParams(qasm=circuit.qasm)
    self.input = circuit
analyze(self) async

Runs the circuit analysis.

Returns:

Type Description
Analysis

The analysis result.

Source code in classiq/analyzer/analyzer.py
async def analyze_async(self) -> analysis_result.Analysis:
    """Runs the circuit analysis.

    Returns:
        The analysis result.
    """
    result = await ApiWrapper.call_analysis_task(self._params)

    if result.status != analysis_result.AnalysisStatus.SUCCESS:
        raise ClassiqAnalyzerError(f"Analysis failed: {result.details}")
    details = validate_type(
        obj=result.details,
        expected_type=analysis_result.Analysis,
        operation="Analysis",
        exception_type=ClassiqAnalyzerError,
    )

    dashboard_path = routes.ANALYZER_DASHBOARD
    self._open_route(path=dashboard_path)
    return details
analyze_async(self) async

Runs the circuit analysis.

Returns:

Type Description
Analysis

The analysis result.

Source code in classiq/analyzer/analyzer.py
async def analyze_async(self) -> analysis_result.Analysis:
    """Runs the circuit analysis.

    Returns:
        The analysis result.
    """
    result = await ApiWrapper.call_analysis_task(self._params)

    if result.status != analysis_result.AnalysisStatus.SUCCESS:
        raise ClassiqAnalyzerError(f"Analysis failed: {result.details}")
    details = validate_type(
        obj=result.details,
        expected_type=analysis_result.Analysis,
        operation="Analysis",
        exception_type=ClassiqAnalyzerError,
    )

    dashboard_path = routes.ANALYZER_DASHBOARD
    self._open_route(path=dashboard_path)
    return details
analyzer_app(self) async

Opens the analyzer app with synthesis interactive results.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def analyzer_app_async(self) -> None:
    """Opens the analyzer app with synthesis interactive results.

    Returns:
        None.
    """
    result = await ApiWrapper.call_analyzer_app(self.input)
    webbrowser.open_new_tab(urljoin(routes.ANALYZER_FULL_FE_APP, str(result.id)))
analyzer_app_async(self) async

Opens the analyzer app with synthesis interactive results.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def analyzer_app_async(self) -> None:
    """Opens the analyzer app with synthesis interactive results.

    Returns:
        None.
    """
    result = await ApiWrapper.call_analyzer_app(self.input)
    webbrowser.open_new_tab(urljoin(routes.ANALYZER_FULL_FE_APP, str(result.id)))
get_hardware_comparison_table(self, hardware=None) async

create a comparison table between the transpiled circuits result on different hardware. The comparison table included the depth, multi qubit gates count,and total gates count of the circuits.

Args: hardware (): List of hardware (string or AnalyzerProviderVendor). if None, the table will include all the available hardware.

Returns: None.

Source code in classiq/analyzer/analyzer.py
async def get_hardware_comparison_table_async(
    self,
    hardware: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
) -> None:
    """create a comparison table between the transpiled circuits result on different hardware.
    The  comparison table included the depth, multi qubit gates count,and total gates count of the circuits.

    Args: hardware (): List of hardware (string or `AnalyzerProviderVendor`). if None, the table will include all
    the available hardware.

    Returns: None.
    """
    if hardware is None:
        hardware = list(AnalyzerProviderVendor)
    params = analysis_params.AnalysisTableParams(
        qasm=self._params.qasm, hardware=hardware
    )
    result = await ApiWrapper.call_table_graphs_task(params=params)
    self.hardware_comparison_table = go.Figure(json.loads(result.graph))
get_hardware_comparison_table_async(self, hardware=None) async

create a comparison table between the transpiled circuits result on different hardware. The comparison table included the depth, multi qubit gates count,and total gates count of the circuits.

Args: hardware (): List of hardware (string or AnalyzerProviderVendor). if None, the table will include all the available hardware.

Returns: None.

Source code in classiq/analyzer/analyzer.py
async def get_hardware_comparison_table_async(
    self,
    hardware: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
) -> None:
    """create a comparison table between the transpiled circuits result on different hardware.
    The  comparison table included the depth, multi qubit gates count,and total gates count of the circuits.

    Args: hardware (): List of hardware (string or `AnalyzerProviderVendor`). if None, the table will include all
    the available hardware.

    Returns: None.
    """
    if hardware is None:
        hardware = list(AnalyzerProviderVendor)
    params = analysis_params.AnalysisTableParams(
        qasm=self._params.qasm, hardware=hardware
    )
    result = await ApiWrapper.call_table_graphs_task(params=params)
    self.hardware_comparison_table = go.Figure(json.loads(result.graph))
get_qubits_connectivity(self) async

create a network connectivity graph of the analysed circuit.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def get_qubits_connectivity_async(self) -> None:
    """create a network connectivity graph of the analysed circuit.

    Returns:
        None.
    """
    result = await ApiWrapper.call_graphs_task(self._params)
    self.graph = go.Figure(json.loads(result.graph))
get_qubits_connectivity_async(self) async

create a network connectivity graph of the analysed circuit.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def get_qubits_connectivity_async(self) -> None:
    """create a network connectivity graph of the analysed circuit.

    Returns:
        None.
    """
    result = await ApiWrapper.call_graphs_task(self._params)
    self.graph = go.Figure(json.loads(result.graph))
plot_hardware_comparison_table(self) async

plot the comparison table. if it has not been created it, it first creates the table using all the available hardware.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_hardware_comparison_table_async(
    self,
) -> None:
    """plot the comparison table. if it has not been created it, it first creates the table using all the
    available hardware.

    Returns:
        None.
    """
    if not hasattr(self, "hardware_comparison_table"):
        await self.get_hardware_comparison_table_async()
    self.hardware_comparison_table.show()
plot_hardware_comparison_table_async(self) async

plot the comparison table. if it has not been created it, it first creates the table using all the available hardware.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_hardware_comparison_table_async(
    self,
) -> None:
    """plot the comparison table. if it has not been created it, it first creates the table using all the
    available hardware.

    Returns:
        None.
    """
    if not hasattr(self, "hardware_comparison_table"):
        await self.get_hardware_comparison_table_async()
    self.hardware_comparison_table.show()
plot_qubits_connectivity(self) async

plot the connectivity graph. if it has not been created it, it first creates the graph.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_qubits_connectivity_async(self) -> None:
    """plot the connectivity graph. if it has not been created it, it first creates the graph.

    Returns:
        None.
    """
    if not hasattr(self, "graph"):
        await self.get_qubits_connectivity_async()
    self.graph.show()
plot_qubits_connectivity_async(self) async

plot the connectivity graph. if it has not been created it, it first creates the graph.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_qubits_connectivity_async(self) -> None:
    """plot the connectivity graph. if it has not been created it, it first creates the graph.

    Returns:
        None.
    """
    if not hasattr(self, "graph"):
        await self.get_qubits_connectivity_async()
    self.graph.show()

rb

RBAnalysis

Source code in classiq/analyzer/rb.py
class RBAnalysis(metaclass=Asyncify):
    def __init__(self, experiments_data: List[AnalysisRBParams]):
        """Init self.

        Args:
            experiments_data: List of results from varius RB experiments.
        """

        self.experiments_data = experiments_data
        self._total_results: pd.DataFrame = pd.DataFrame()

    async def _get_multiple_hardware_results_async(self) -> Dict[str, RbResults]:
        total_result: Dict[str, RbResults] = {}
        for batch in self.experiments_data:
            if len(batch.num_clifford) < 5:
                raise ClassiqAnalyzerError(
                    f"An experiment mush contain at least five sequences,"
                    f" this sequence is {len(batch.num_clifford)}"
                )
            rb_result = await ApiWrapper.call_rb_analysis_task(batch)
            total_result[batch.hardware] = rb_result
        return total_result

    @staticmethod
    def _get_df_indices(results) -> List[str]:
        temp_res = results.copy()
        _, rb_result_keys = temp_res.popitem()
        return list(rb_result_keys.__dict__.keys())

    async def show_multiple_hardware_data_async(self) -> pd.DataFrame:
        """Run the RB analysis.

        Returns:
            The RB result.
        """
        results = await self._get_multiple_hardware_results_async()
        indices = RBAnalysis._get_df_indices(results)
        result_df = pd.DataFrame(index=indices)
        for hardware, result in results.items():
            result_df[hardware] = result.__dict__.values()
        self._total_results = result_df
        return result_df

    def plot_multiple_hardware_results(self) -> go.Figure:
        """Plot Bar graph of the results.

        Returns:
            None.
        """
        df = self._total_results.loc[["mean_fidelity", "average_error"]].transpose()
        hardware = list(df.index)
        params = list(df.columns)
        data = []
        for param in params:
            data.append(go.Bar(name=param, x=hardware, y=df[param].values * 100))
        fig = go.Figure(data).update_layout(
            title="RB hardware comparison",
            barmode="group",
            yaxis=dict(title="Fidelity in %"),
            xaxis=dict(title="Hardware"),
        )
        return fig
__init__(self, experiments_data) special

Init self.

Parameters:

Name Type Description Default
experiments_data List[classiq.interface.analyzer.analysis_params.AnalysisRBParams]

List of results from varius RB experiments.

required
Source code in classiq/analyzer/rb.py
def __init__(self, experiments_data: List[AnalysisRBParams]):
    """Init self.

    Args:
        experiments_data: List of results from varius RB experiments.
    """

    self.experiments_data = experiments_data
    self._total_results: pd.DataFrame = pd.DataFrame()
plot_multiple_hardware_results(self)

Plot Bar graph of the results.

Returns:

Type Description
Figure

None.

Source code in classiq/analyzer/rb.py
def plot_multiple_hardware_results(self) -> go.Figure:
    """Plot Bar graph of the results.

    Returns:
        None.
    """
    df = self._total_results.loc[["mean_fidelity", "average_error"]].transpose()
    hardware = list(df.index)
    params = list(df.columns)
    data = []
    for param in params:
        data.append(go.Bar(name=param, x=hardware, y=df[param].values * 100))
    fig = go.Figure(data).update_layout(
        title="RB hardware comparison",
        barmode="group",
        yaxis=dict(title="Fidelity in %"),
        xaxis=dict(title="Hardware"),
    )
    return fig
show_multiple_hardware_data(self) async

Run the RB analysis.

Returns:

Type Description
DataFrame

The RB result.

Source code in classiq/analyzer/rb.py
async def show_multiple_hardware_data_async(self) -> pd.DataFrame:
    """Run the RB analysis.

    Returns:
        The RB result.
    """
    results = await self._get_multiple_hardware_results_async()
    indices = RBAnalysis._get_df_indices(results)
    result_df = pd.DataFrame(index=indices)
    for hardware, result in results.items():
        result_df[hardware] = result.__dict__.values()
    self._total_results = result_df
    return result_df
show_multiple_hardware_data_async(self) async

Run the RB analysis.

Returns:

Type Description
DataFrame

The RB result.

Source code in classiq/analyzer/rb.py
async def show_multiple_hardware_data_async(self) -> pd.DataFrame:
    """Run the RB analysis.

    Returns:
        The RB result.
    """
    results = await self._get_multiple_hardware_results_async()
    indices = RBAnalysis._get_df_indices(results)
    result_df = pd.DataFrame(index=indices)
    for hardware, result in results.items():
        result_df[hardware] = result.__dict__.values()
    self._total_results = result_df
    return result_df

executor

Executor module, implementing facilities for executing quantum programs using Classiq platform.

Executor

Executor is the entry point for executing quantum programs on multiple quantum hardware vendors.

Source code in classiq/executor.py
class Executor(metaclass=Asyncify):
    """Executor is the entry point for executing quantum programs on multiple quantum hardware vendors."""

    def __init__(
        self, preferences: Optional[ExecutionPreferences] = None, **kwargs
    ) -> None:
        """Init self.

        Args:
            preferences (): Execution preferences, such as number of shots.
        """
        self._preferences = preferences or ExecutionPreferences(**kwargs)

    @property
    def preferences(self) -> ExecutionPreferences:
        return self._preferences

    async def _execute(self, expected_type: Any, **execution_request_kwargs):
        request = execution_request.ExecutionRequest(
            preferences=self._preferences, **execution_request_kwargs
        )

        try:
            execution_result = await ApiWrapper.call_execute_task(request=request)
        except Exception as exc:
            raise ClassiqExecutionError(f"Execution failed: {exc!s}") from exc

        if execution_result.status != exc_result.ExecutionStatus.SUCCESS:
            raise ClassiqExecutionError(f"Execution failed: {execution_result.details}")

        return validate_type(
            obj=execution_result.details,
            expected_type=expected_type,
            operation="Execution",
            exception_type=ClassiqExecutionError,
        )

    async def execute_quantum_program_async(
        self,
        quantum_program: QuantumProgram,
        arguments: Optional[Arguments] = None,
    ) -> ExecutionDetails:
        kwargs = quantum_program.dict()
        if arguments:
            original_arguments = kwargs["arguments"] or {}
            kwargs["arguments"] = {**original_arguments, **arguments}

        return await self._execute(
            execution_payload=execution_request.QuantumProgramExecution(**kwargs),
            expected_type=ExecutionDetails,
        )

    async def batch_execute_quantum_program_async(
        self, quantum_programs: Collection[QuantumProgram]
    ) -> List[ProgramAndResult]:
        results = await asyncio.gather(
            *map(self.execute_quantum_program_async, quantum_programs),
            return_exceptions=True,
        )
        return list(zip(quantum_programs, results))

    async def execute_generated_circuit_async(
        self, generation_result: GeneratedCircuit
    ) -> Union[FinanceSimulationResults, GroverSimulationResults]:
        if generation_result.metadata is None:
            raise ClassiqExecutionError(
                "The execute_generated_circuit is to execute generated circuits as oracles, but "
                "the generated circuit's metadata is empty. To execute a circuit as-is, please"
                "use execute_quantum_program."
            )

        return await self._execute(
            execution_payload=execution_request.GenerationMetadataExecution(
                **generation_result.metadata.dict()
            ),
            expected_type=(FinanceSimulationResults, GroverSimulationResults),
        )

    async def execute_hamiltonian_minimization_async(
        self,
        hamiltonian_minimization_problem: HamiltonianMinimizationProblem,
    ) -> VQESolverResult:
        return await self._execute(
            execution_payload=execution_request.HamiltonianMinimizationProblemExecution(
                **hamiltonian_minimization_problem.dict()
            ),
            expected_type=VQESolverResult,
        )

    async def execute_async(
        self,
        arg: Union[GeneratedCircuit, QuantumProgram, str],
        *args,
        **kwargs,
    ):
        if isinstance(arg, GeneratedCircuit):
            if arg.metadata is not None:
                return await self.execute_generated_circuit_async(
                    arg,
                    *args,
                    **kwargs,
                )
            else:
                first_format, first_output = next(iter(arg.outputs.items()))
                program = QuantumProgram(code=first_output, syntax=first_format)
        elif isinstance(arg, QuantumProgram):
            program = arg
        elif isinstance(arg, str):
            program = QuantumProgram(code=arg)
        else:
            raise ClassiqValueError("Invalid executor input")

        return await self.execute_quantum_program_async(
            program,
            *args,
            **kwargs,
        )

__init__(self, preferences=None, **kwargs) special

Init self.

Parameters:

Name Type Description Default
preferences

Execution preferences, such as number of shots.

None
Source code in classiq/executor.py
def __init__(
    self, preferences: Optional[ExecutionPreferences] = None, **kwargs
) -> None:
    """Init self.

    Args:
        preferences (): Execution preferences, such as number of shots.
    """
    self._preferences = preferences or ExecutionPreferences(**kwargs)

batch_execute_multiple_backends(preferences_template, backend_preferences, quantum_programs) async

Execute all the provided quantum programs (n) on all the provided backends (m). In total, m * n executions. The return value is a list of the following tuples:

  • An element from backend_preferences
  • An element from quantum_programs
  • The execution result of the quantum program on the backend. If the execution failed, the value is an exception.

The length of the list is m * n.

The preferences_template argument is used to supplement all other preferences.

The code is equivalent to:

for backend in backend_preferences:
    for program in quantum_programs:
        preferences = preferences_template.copy()
        preferences.backend_preferences = backend
        Executor(preferences).execute_quantum_program(program)

Source code in classiq/executor.py
async def batch_execute_multiple_backends_async(
    preferences_template: ExecutionPreferences,
    backend_preferences: Sequence[BackendPreferencesTypes],
    quantum_programs: Collection[QuantumProgram],
) -> List[BackendPreferencesProgramAndResult]:
    """
    Execute all the provided quantum programs (n) on all the provided backends (m).
    In total, m * n executions.
    The return value is a list of the following tuples:

    - An element from `backend_preferences`
    - An element from `quantum_programs`
    - The execution result of the quantum program on the backend. If the execution failed,
      the value is an exception.

    The length of the list is m * n.

    The `preferences_template` argument is used to supplement all other preferences.

    The code is equivalent to:
    ```
    for backend in backend_preferences:
        for program in quantum_programs:
            preferences = preferences_template.copy()
            preferences.backend_preferences = backend
            Executor(preferences).execute_quantum_program(program)
    ```
    """
    executors = [
        Executor(
            preferences=preferences_template.copy(
                update={"backend_preferences": backend}
            )
        )
        for backend in backend_preferences
    ]
    results = await asyncio.gather(
        *(
            executor.batch_execute_quantum_program_async(quantum_programs)
            for executor in executors
        ),
        return_exceptions=True,
    )

    def map_return_value(
        executor: Executor,
        result: Union[List[ProgramAndResult], BaseException],
    ) -> Iterable[BackendPreferencesProgramAndResult]:
        nonlocal quantum_programs
        preferences = executor.preferences.backend_preferences
        if isinstance(result, BaseException):
            return ((preferences, program, result) for program in quantum_programs)
        else:
            return (
                (preferences, program, single_result)
                for program, single_result in result
            )

    return list(
        itertools.chain.from_iterable(
            map_return_value(executor, result)
            for executor, result in zip(executors, results)
        )
    )

batch_execute_multiple_backends_async(preferences_template, backend_preferences, quantum_programs) async

Execute all the provided quantum programs (n) on all the provided backends (m). In total, m * n executions. The return value is a list of the following tuples:

  • An element from backend_preferences
  • An element from quantum_programs
  • The execution result of the quantum program on the backend. If the execution failed, the value is an exception.

The length of the list is m * n.

The preferences_template argument is used to supplement all other preferences.

The code is equivalent to:

for backend in backend_preferences:
    for program in quantum_programs:
        preferences = preferences_template.copy()
        preferences.backend_preferences = backend
        Executor(preferences).execute_quantum_program(program)

Source code in classiq/executor.py
async def batch_execute_multiple_backends_async(
    preferences_template: ExecutionPreferences,
    backend_preferences: Sequence[BackendPreferencesTypes],
    quantum_programs: Collection[QuantumProgram],
) -> List[BackendPreferencesProgramAndResult]:
    """
    Execute all the provided quantum programs (n) on all the provided backends (m).
    In total, m * n executions.
    The return value is a list of the following tuples:

    - An element from `backend_preferences`
    - An element from `quantum_programs`
    - The execution result of the quantum program on the backend. If the execution failed,
      the value is an exception.

    The length of the list is m * n.

    The `preferences_template` argument is used to supplement all other preferences.

    The code is equivalent to:
    ```
    for backend in backend_preferences:
        for program in quantum_programs:
            preferences = preferences_template.copy()
            preferences.backend_preferences = backend
            Executor(preferences).execute_quantum_program(program)
    ```
    """
    executors = [
        Executor(
            preferences=preferences_template.copy(
                update={"backend_preferences": backend}
            )
        )
        for backend in backend_preferences
    ]
    results = await asyncio.gather(
        *(
            executor.batch_execute_quantum_program_async(quantum_programs)
            for executor in executors
        ),
        return_exceptions=True,
    )

    def map_return_value(
        executor: Executor,
        result: Union[List[ProgramAndResult], BaseException],
    ) -> Iterable[BackendPreferencesProgramAndResult]:
        nonlocal quantum_programs
        preferences = executor.preferences.backend_preferences
        if isinstance(result, BaseException):
            return ((preferences, program, result) for program in quantum_programs)
        else:
            return (
                (preferences, program, single_result)
                for program, single_result in result
            )

    return list(
        itertools.chain.from_iterable(
            map_return_value(executor, result)
            for executor, result in zip(executors, results)
        )
    )

interface special

jobs

JobStatus (str, Enum)

An enumeration.

Source code in classiq/interface/jobs.py
class JobStatus(str, Enum):
    QUEUED = "QUEUED"
    RUNNING = "RUNNING"
    READY = "READY"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"
    CANCELLING = "CANCELLING"
    CANCELLED = "CANCELLED"
    UNKNOWN = "UNKNOWN"

    def is_final(self) -> bool:
        return self in (self.COMPLETED, self.FAILED, self.CANCELLED)

status

Status (str, Enum)

An enumeration.

Source code in classiq/interface/status.py
class Status(str, enum.Enum):
    SUCCESS = "SUCCESS"
    ERROR = "ERROR"

model_designer special

function_handler

FunctionHandler (ABC)

Source code in classiq/model_designer/function_handler.py
class FunctionHandler(abc.ABC):
    def __init__(self) -> None:
        self._generated_wires: Set[Wire] = set()
        self._function_library: Optional[FunctionLibrary] = None

    def _verify_legal_wires(self, wires: WireOrWires) -> None:
        if isinstance(wires, Wire):
            wires = [wires]
        if not all(wire in self._generated_wires for wire in wires):
            raise ClassiqWiringError("Wire does not belong to this generator")

    def _update_generated_wires(self, wires: WireOrWires) -> None:
        if isinstance(wires, Wire):
            wires = [wires]
        self._generated_wires.update(wires)

    def apply(
        self,
        function_name: str,
        in_wires: Optional[Union[Dict[str, QregOrWire], QuantumRegister]] = None,
        out_wires: Optional[Union[Dict[str, QregOrWire], QuantumRegister]] = None,
        is_inverse: bool = False,
        control_state: Optional[ControlState] = None,
        call_name: Optional[str] = None,
    ) -> Dict[str, Wire]:
        if self._function_library is None:
            raise ClassiqValueError("Cannot apply function without a function library")

        params = self._function_library.get_function(function_name)
        return self._function_call_handler(
            CustomFunction.__name__,
            params,
            in_wires=in_wires,
            out_wires=out_wires,
            is_inverse=is_inverse,
            control_state=control_state,
            call_name=call_name,
        )

    def _function_call_handler(
        self,
        function: str,
        params: function_params.FunctionParams,
        in_wires: Optional[Union[Dict[str, QregOrWire], QuantumRegister]] = None,
        out_wires: Optional[Union[Dict[str, QregOrWire], QuantumRegister]] = None,
        is_inverse: bool = False,
        control_state: Optional[ControlState] = None,
        call_name: Optional[str] = None,
    ) -> Dict[str, Wire]:
        if function != type(params).__name__:
            raise ClassiqValueError(
                "The FunctionParams type does not match function name"
            )

        if isinstance(params, CustomFunction):
            FunctionLibraryData.validate_function_in_library(
                library=self._function_library.data if self._function_library else None,
                function_params=params,
                error_handler=ClassiqValueError,
            )

        call = function_call.FunctionCall(
            function=function,
            function_params=params,
            is_inverse=is_inverse,
            control_state=control_state,
            name=call_name,
        )

        if in_wires is not None:
            self._connect_in_wires(call=call, in_wires=in_wires)

        self._logic_flow.append(call)

        return self._connect_out_wires(
            call=call,
            out_wires=out_wires or {},
        )

    def _connect_in_wires(
        self,
        call: function_call.FunctionCall,
        in_wires: SupportedIOUnion,
    ) -> None:
        if isinstance(in_wires, dict):
            self._connect_named_in_wires(call=call, in_wires=in_wires)
        elif isinstance(in_wires, QuantumRegister):
            self._connect_unnamed_in_quantum_registers(
                call=call, quantum_registers=[in_wires]
            )
        elif isinstance(in_wires, collections.abc.Collection):
            self._connect_unnamed_in_quantum_registers(
                # mypy doesn't recognize that `dict` wouldn't reach this point
                call=call,
                quantum_registers=in_wires,  # type: ignore[arg-type]
            )
        else:
            raise ClassiqWiringError(
                f"Invalid in_wires type: {type(in_wires).__name__}"
            )

    def _connect_unnamed_in_quantum_registers(
        self,
        call: function_call.FunctionCall,
        quantum_registers: Collection[QuantumRegister],
    ) -> None:
        self._verify_legal_wires((i.wire for i in quantum_registers))

        call_inputs = call.function_params.get_io_names(
            function_params.IO.Input, call.is_inverse, call.control_state
        )

        if len(call_inputs) != len(quantum_registers):
            raise ClassiqWiringError(
                f'A call to "{call.name}" requires {len(call_inputs)} items, but {len(quantum_registers)} were given'
            )

        for input_name, quantum_register in zip(call_inputs, quantum_registers):
            quantum_register.consume()
            quantum_register.wire.connect_wire_end(end_call=call, input_name=input_name)

    def _connect_named_in_wires(
        self, call: function_call.FunctionCall, in_wires: Dict[str, QregOrWire]
    ) -> None:
        self._verify_legal_wires(
            [
                in_wire.wire if isinstance(in_wire, QuantumRegister) else in_wire
                for in_wire in in_wires.values()
            ]
        )

        for input_name, in_wire in in_wires.items():
            if isinstance(in_wire, QuantumRegister):
                in_wire.consume()
                in_wire = in_wire.wire

            in_wire.connect_wire_end(end_call=call, input_name=input_name)

    def _connect_out_wires(
        self,
        call: function_call.FunctionCall,
        out_wires: SupportedIOUnion,
    ) -> Dict[str, Wire]:
        if isinstance(out_wires, dict):
            wire_dict = self._connect_named_out_wires(call=call, out_wires=out_wires)
        elif isinstance(out_wires, QuantumRegister):
            wire_dict = self._connect_unnamed_out_quantum_registers(
                call=call, quantum_registers=[out_wires]
            )
        elif isinstance(out_wires, collections.abc.Collection):
            if not all(isinstance(i, QuantumRegister) for i in out_wires):
                raise ClassiqWiringError(
                    "When supplying and iterable, all items must be instances of QReg"
                )
            wire_dict = self._connect_unnamed_out_quantum_registers(
                call=call, quantum_registers=out_wires  # type: ignore[arg-type]
            )
        else:
            raise ClassiqWiringError(
                f"Invalid out_wires type: {type(out_wires).__name__}"
            )

        self._update_generated_wires(wire_dict.values())
        return wire_dict

    def _connect_named_out_wires(
        self,
        call: function_call.FunctionCall,
        out_wires: Dict[str, QregOrWire],
    ) -> Dict[str, Wire]:
        wire_dict: Dict[str, Wire] = {}
        output_names = call.function_params.get_io_names(
            function_params.IO.Output, call.is_inverse, call.control_state
        )

        specified_outputs: List[Tuple[str, str]] = list(tuple())
        for specified_output in out_wires.keys():
            match: Optional[Match] = re.fullmatch(
                function_call.IO_REGEX, specified_output
            )
            if match is None:
                raise ClassiqWiringError(
                    f"Output ({specified_output}) is not a valid expression"
                )
            name = match.groupdict().get(function_call.NAME)
            slicing = match.groupdict().get(function_call.SLICING)
            if name is None or name not in output_names:
                raise ClassiqWiringError(
                    f"Output name ({name}) does not belong to this function call"
                )
            if (
                slicing is not None
                and re.fullmatch(function_call.LEGAL_SLICING, slicing) is None
            ):
                raise ClassiqWiringError(
                    f"Slicing / indexing expression ({slicing}) is illegal"
                )

            specified_outputs.append(
                (name, f"[{slicing}]" if slicing is not None else "")
            )

        for output_name in output_names:
            connected_outputs = [
                name + slicing
                for name, slicing in specified_outputs
                if output_name == name
            ]

            if not connected_outputs:
                wire_dict[output_name] = self._output_wire_type(
                    start_call=call, output_name=output_name
                )
                continue

            for specified_output in connected_outputs:
                out_wire = out_wires[specified_output]
                if isinstance(out_wire, QuantumRegister):
                    out_wire = out_wire.wire

                out_wire.connect_wire_start(
                    start_call=call, output_name=specified_output
                )
                wire_dict[specified_output] = out_wire

        return wire_dict

    def _connect_unnamed_out_quantum_registers(
        self,
        call: function_call.FunctionCall,
        quantum_registers: Collection[QuantumRegister],
    ) -> Dict[str, Wire]:
        wire_dict: Dict[str, Wire] = {}
        output_names = call.function_params.get_io_names(
            function_params.IO.Output, call.is_inverse, call.control_state
        )

        for quantum_register, output_name in zip(quantum_registers, output_names):
            quantum_register.wire.connect_wire_start(
                start_call=call, output_name=output_name
            )
            wire_dict[output_name] = quantum_register.wire

        return wire_dict

    def __getattr__(self, item):
        is_builtin_function_name = any(
            item == func.__name__
            for func in function_param_list.get_function_param_list()
        )

        if is_builtin_function_name:
            return functools.partial(self._function_call_handler, item)

        is_user_function_name = (
            self._function_library is not None
            and item in self._function_library.function_names
        )

        if is_user_function_name:
            return functools.partial(self.apply, item)

        raise AttributeError(f"'{self.__class__.__name__}' has no attribute '{item}'")

    def __dir__(self):
        builtin_func_name = [
            func.__name__ for func in function_param_list.get_function_param_list()
        ]
        user_func_names = (
            list(self._function_library.function_names)
            if self._function_library is not None
            else list()
        )
        return list(super().__dir__()) + builtin_func_name + user_func_names

    def include_library(self, library: FunctionLibrary) -> None:
        """Includes a function library.

        Args:
            library (FunctionLibrary): The function library.
        """
        if self._function_library is not None:
            raise ClassiqValueError("Another function library is already included.")

        self._function_library = library

    @property
    @abc.abstractmethod
    def _logic_flow(self) -> List[function_call.FunctionCall]:
        pass

    @property
    @abc.abstractmethod
    def _output_wire_type(self) -> Type[Wire]:
        pass
include_library(self, library)

Includes a function library.

Parameters:

Name Type Description Default
library FunctionLibrary

The function library.

required
Source code in classiq/model_designer/function_handler.py
def include_library(self, library: FunctionLibrary) -> None:
    """Includes a function library.

    Args:
        library (FunctionLibrary): The function library.
    """
    if self._function_library is not None:
        raise ClassiqValueError("Another function library is already included.")

    self._function_library = library

model_designer

ModelDesigner module, implementing facilities for designing models and generating circuits using Classiq platform.

ModelDesigner (FunctionHandler)

Facility to generate circuits, based on the model.

Source code in classiq/model_designer/model_designer.py
class ModelDesigner(function_handler.FunctionHandler, metaclass=AsyncifyABC):
    """Facility to generate circuits, based on the model."""

    def __init__(self, **kwargs) -> None:
        """Init self."""
        super().__init__()
        self._model = Model(**kwargs)

    @property
    def _output_wire_type(self) -> Type[wire.Wire]:
        return wire.Wire

    @property
    def _logic_flow(self) -> List[function_call.FunctionCall]:
        return self._model.logic_flow

    @property
    def constraints(self) -> Constraints:
        """Get the constraints aggregated in self.

        Returns:
            The constraints data.
        """
        return self._model.constraints

    @property
    def preferences(self) -> Preferences:
        """Get the preferences aggregated in self.

        Returns:
            The preferences data.
        """
        return self._model.preferences

    async def synthesize_async(
        self,
        constraints: Optional[Constraints] = None,
        preferences: Optional[Preferences] = None,
    ) -> result.GeneratedCircuit:
        """Async version of `generate`
        Generates a circuit, based on the aggregation of requirements in self.

        Returns:
            The results of the generation procedure.
        """
        self._model.preferences = preferences or self._model.preferences
        self._model.constraints = constraints or self._model.constraints
        try:
            generation_result = await ApiWrapper.call_generation_task(self._model)
        except ClassiqAPIError as exc:
            raise ClassiqGenerationError(f"Generation failed: {exc}") from exc

        if (
            generation_result.status != result.GenerationStatus.SUCCESS
            or not isinstance(generation_result.details, result.GeneratedCircuit)
        ):
            raise ClassiqGenerationError(
                f"Generation failed: {generation_result.details}"
            )

        generation_result.details.model = self._model
        return generation_result.details

    def include_library(self, library: FunctionLibrary) -> None:
        """Includes a user-defined custom function library.

        Args:
            library (FunctionLibrary): The custom function library.
        """
        super().include_library(library=library)
        self._model.function_library = library.data

    def dumps(self, ignore_warning: bool = False) -> str:
        """Serialize model to a JSON formatted `str`

        Args:
            ignore_warning (bool): Whether to ignore the warning print
        """
        if not ignore_warning:
            _logger.warning(
                "Saving to json is currently unstable since versions may change"
            )

        return self._model.json(exclude_defaults=True, indent=4)

    def dump(
        self, fp: Optional[_SupportedIO] = None, ignore_warning: bool = False
    ) -> None:
        """Serialize model to a JSON formatted stream to `fp` (a `.write()`)-supporting file-like object

        Args:
            fp (IO | str | None): a file-like object
                if None -> a temporaty file will be created
                if str -> this will be treated as the file path
            ignore_warning (bool): Whether to ignore the warning print
        """
        with _file_handler(fp, "w") as f:
            f.write(self.dumps(ignore_warning=ignore_warning))

    @classmethod
    def loads(cls, s: AnyStr) -> ModelDesigner:
        """Deserialize `s`, a JSON formatted `str`, to a ModelDesigner

        Args:
            s (str | bytes): A JSON-formatted `str` | `bytes`
        """
        new_instance = cls()
        new_instance._model = Model.parse_raw(s)
        return new_instance

    @classmethod
    def load(cls, fp: Optional[_SupportedIO]) -> ModelDesigner:
        """Deserialize `fp` (a `.read()`-supporting file-like object) containing a JSON formatted document to a ModelDesigner

        Args:
            fp (IO | str): a file-like object
                if str -> this will be treated as the file path
        """
        with _file_handler(fp, "r") as f:
            return cls.loads(f.read())

    def export_to_vscode(self, ignore_warning: bool = False) -> None:
        """Export the model to a file, and display in VisualStudioCode"""

        if not distutils.spawn.find_executable("code"):
            raise ClassiqFileNotFoundError(
                "Please install VSCode to path\nIn VSCode, press [Ctrl/Command]+Shift+p, and then type \"install 'code' command in PATH\""
            )

        fp = tempfile.NamedTemporaryFile("w", suffix=".qmod", delete=False)
        self.dump(fp, ignore_warning=ignore_warning)
        fp.close()
        distutils.spawn.spawn(["code", fp.name])
constraints: Constraints property readonly

Get the constraints aggregated in self.

Returns:

Type Description
Constraints

The constraints data.

preferences: Preferences property readonly

Get the preferences aggregated in self.

Returns:

Type Description
Preferences

The preferences data.

__init__(self, **kwargs) special

Init self.

Source code in classiq/model_designer/model_designer.py
def __init__(self, **kwargs) -> None:
    """Init self."""
    super().__init__()
    self._model = Model(**kwargs)
dump(self, fp=None, ignore_warning=False)

Serialize model to a JSON formatted stream to fp (a .write())-supporting file-like object

Parameters:

Name Type Description Default
fp IO | str | None

a file-like object if None -> a temporaty file will be created if str -> this will be treated as the file path

None
ignore_warning bool

Whether to ignore the warning print

False
Source code in classiq/model_designer/model_designer.py
def dump(
    self, fp: Optional[_SupportedIO] = None, ignore_warning: bool = False
) -> None:
    """Serialize model to a JSON formatted stream to `fp` (a `.write()`)-supporting file-like object

    Args:
        fp (IO | str | None): a file-like object
            if None -> a temporaty file will be created
            if str -> this will be treated as the file path
        ignore_warning (bool): Whether to ignore the warning print
    """
    with _file_handler(fp, "w") as f:
        f.write(self.dumps(ignore_warning=ignore_warning))
dumps(self, ignore_warning=False)

Serialize model to a JSON formatted str

Parameters:

Name Type Description Default
ignore_warning bool

Whether to ignore the warning print

False
Source code in classiq/model_designer/model_designer.py
def dumps(self, ignore_warning: bool = False) -> str:
    """Serialize model to a JSON formatted `str`

    Args:
        ignore_warning (bool): Whether to ignore the warning print
    """
    if not ignore_warning:
        _logger.warning(
            "Saving to json is currently unstable since versions may change"
        )

    return self._model.json(exclude_defaults=True, indent=4)
export_to_vscode(self, ignore_warning=False)

Export the model to a file, and display in VisualStudioCode

Source code in classiq/model_designer/model_designer.py
def export_to_vscode(self, ignore_warning: bool = False) -> None:
    """Export the model to a file, and display in VisualStudioCode"""

    if not distutils.spawn.find_executable("code"):
        raise ClassiqFileNotFoundError(
            "Please install VSCode to path\nIn VSCode, press [Ctrl/Command]+Shift+p, and then type \"install 'code' command in PATH\""
        )

    fp = tempfile.NamedTemporaryFile("w", suffix=".qmod", delete=False)
    self.dump(fp, ignore_warning=ignore_warning)
    fp.close()
    distutils.spawn.spawn(["code", fp.name])
include_library(self, library)

Includes a user-defined custom function library.

Parameters:

Name Type Description Default
library FunctionLibrary

The custom function library.

required
Source code in classiq/model_designer/model_designer.py
def include_library(self, library: FunctionLibrary) -> None:
    """Includes a user-defined custom function library.

    Args:
        library (FunctionLibrary): The custom function library.
    """
    super().include_library(library=library)
    self._model.function_library = library.data
load(fp) classmethod

Deserialize fp (a .read()-supporting file-like object) containing a JSON formatted document to a ModelDesigner

Parameters:

Name Type Description Default
fp IO | str

a file-like object if str -> this will be treated as the file path

required
Source code in classiq/model_designer/model_designer.py
@classmethod
def load(cls, fp: Optional[_SupportedIO]) -> ModelDesigner:
    """Deserialize `fp` (a `.read()`-supporting file-like object) containing a JSON formatted document to a ModelDesigner

    Args:
        fp (IO | str): a file-like object
            if str -> this will be treated as the file path
    """
    with _file_handler(fp, "r") as f:
        return cls.loads(f.read())
loads(s) classmethod

Deserialize s, a JSON formatted str, to a ModelDesigner

Parameters:

Name Type Description Default
s str | bytes

A JSON-formatted str | bytes

required
Source code in classiq/model_designer/model_designer.py
@classmethod
def loads(cls, s: AnyStr) -> ModelDesigner:
    """Deserialize `s`, a JSON formatted `str`, to a ModelDesigner

    Args:
        s (str | bytes): A JSON-formatted `str` | `bytes`
    """
    new_instance = cls()
    new_instance._model = Model.parse_raw(s)
    return new_instance
synthesize(self, constraints=None, preferences=None) async

Async version of generate Generates a circuit, based on the aggregation of requirements in self.

Returns:

Type Description
result.GeneratedCircuit

The results of the generation procedure.

Source code in classiq/model_designer/model_designer.py
async def synthesize_async(
    self,
    constraints: Optional[Constraints] = None,
    preferences: Optional[Preferences] = None,
) -> result.GeneratedCircuit:
    """Async version of `generate`
    Generates a circuit, based on the aggregation of requirements in self.

    Returns:
        The results of the generation procedure.
    """
    self._model.preferences = preferences or self._model.preferences
    self._model.constraints = constraints or self._model.constraints
    try:
        generation_result = await ApiWrapper.call_generation_task(self._model)
    except ClassiqAPIError as exc:
        raise ClassiqGenerationError(f"Generation failed: {exc}") from exc

    if (
        generation_result.status != result.GenerationStatus.SUCCESS
        or not isinstance(generation_result.details, result.GeneratedCircuit)
    ):
        raise ClassiqGenerationError(
            f"Generation failed: {generation_result.details}"
        )

    generation_result.details.model = self._model
    return generation_result.details
synthesize_async(self, constraints=None, preferences=None) async

Async version of generate Generates a circuit, based on the aggregation of requirements in self.

Returns:

Type Description
result.GeneratedCircuit

The results of the generation procedure.

Source code in classiq/model_designer/model_designer.py
async def synthesize_async(
    self,
    constraints: Optional[Constraints] = None,
    preferences: Optional[Preferences] = None,
) -> result.GeneratedCircuit:
    """Async version of `generate`
    Generates a circuit, based on the aggregation of requirements in self.

    Returns:
        The results of the generation procedure.
    """
    self._model.preferences = preferences or self._model.preferences
    self._model.constraints = constraints or self._model.constraints
    try:
        generation_result = await ApiWrapper.call_generation_task(self._model)
    except ClassiqAPIError as exc:
        raise ClassiqGenerationError(f"Generation failed: {exc}") from exc

    if (
        generation_result.status != result.GenerationStatus.SUCCESS
        or not isinstance(generation_result.details, result.GeneratedCircuit)
    ):
        raise ClassiqGenerationError(
            f"Generation failed: {generation_result.details}"
        )

    generation_result.details.model = self._model
    return generation_result.details

quantum_functions special

annotation_parser

get_annotation_role(annotation, default_role)

!!! note "this function cannot distinguish between inputs and outputs." Thus, for inputs, all 3 options are valid However, for outputs: a) we don't expect to get ZERO b) We treat INPUT as OUTPUT

Source code in classiq/quantum_functions/annotation_parser.py
def get_annotation_role(annotation: GenericAlias, default_role: Role) -> Role:
    """
    Note: this function cannot distinguish between inputs and outputs.
        Thus, for inputs, all 3 options are valid
        However, for outputs:
            a) we don't expect to get ZERO
            b) We treat INPUT as OUTPUT
    """
    ret = None

    if getattr(annotation, "role", None) is not None:
        ret = annotation.role
    if getattr(annotation.__origin__, "role", None) is not None:
        ret = annotation.role

    if issubclass(annotation.__origin__, QReg) and not issubclass(
        annotation.__origin__, ZeroQReg
    ):
        ret = default_role

    if issubclass(annotation.__origin__, ZeroQReg) and not issubclass(
        annotation.__origin__, AuxQReg
    ):
        ret = Role.ZERO

    if issubclass(annotation.__origin__, AuxQReg):
        ret = Role.AUXILIARY

    # Didn't match anything so far
    if ret is None:
        raise ClassiqQFuncError("Invalid annotation role")

    if default_role == Role.INPUT and ret == Role.OUTPUT:
        raise ClassiqQFuncError("input should not have Role.OUTPUT")

    if default_role == Role.OUTPUT and ret in (Role.ZERO, Role.INPUT):
        raise ClassiqQFuncError("output should not have Role.ZERO / Role.INPUT")

    return ret

function_library

Function library module, implementing facilities for adding user defined functions to the Classiq platform.

FunctionLibrary

Facility to manage functions.

Source code in classiq/quantum_functions/function_library.py
class FunctionLibrary:
    """Facility to manage functions."""

    def __init__(self, *functions, name: str = DEFAULT_FUNCTION_LIBRARY_NAME):
        """
        Args:
            name (:obj:`str`, optional): The name of the function library.
            *functions (:obj:`FunctionData`, optional): A list of functions to initialize the object.
        """
        self._data = FunctionLibraryData(name=name)
        self._params: Dict[str, CustomFunction] = dict()

        for f in functions:
            self.add_function(f)

    def get_function(self, function_name: str) -> CustomFunction:
        return self._params[function_name]

    def add_function(
        self,
        function_data: Union[FunctionData, QuantumFunction],
        override_existing_functions: bool = False,
    ) -> CustomFunction:
        """Adds a function to the function library.

        Args:
            function_data (FunctionData): The function data object.
            override_existing_functions (:obj:`bool`, optional): Defaults to False.

        Returns:
            The custom function parameters.
        """
        if isinstance(function_data, QuantumFunction):
            function_data = function_data.function_data

        if not isinstance(function_data, FunctionData):
            raise ClassiqValueError(
                f"FunctionData object expected, got {function_data.__class__.__name__}"
            )

        function_name = function_data.name
        if (
            not override_existing_functions
            and function_name in self._data.function_dict
        ):
            raise ClassiqValueError("Cannot override existing functions.")

        if function_data.function_type == FunctionType.CompositeFunction:
            for call in function_data.logic_flow:
                if not isinstance(call.function_params, CustomFunction):
                    continue
                FunctionLibraryData.validate_function_in_library(
                    library=self._data,
                    function_params=call.function_params,
                    error_handler=ClassiqValueError,
                )

        self._data.function_dict[function_name] = function_data
        self._params[function_name] = self._to_params(function_data)
        return self.get_function(function_name=function_name)

    def remove_function(self, function_name: str) -> FunctionData:
        """Removes a function from the function library.

        Args:
            function_name (str): The name of the function.

        Returns:
            The removed function data.
        """
        self._params.pop(function_name)
        return self._data.function_dict.pop(function_name)

    @property
    def name(self) -> str:
        """The library name."""
        return self._data.name

    @property
    def function_names(self) -> Tuple[str, ...]:
        """Get a tuple of the names of the functions in the library.

        Returns:
            The names of the functions in the library.
        """
        return tuple(self._data.function_dict.keys())

    @property
    def data(self) -> FunctionLibraryData:
        return self._data

    @staticmethod
    def _to_params(data: FunctionData) -> CustomFunction:
        params = CustomFunction(name=data.name)
        params.generate_io_names(
            input_set=data.input_set,
            output_set=data.output_set,
        )
        return params
function_names: Tuple[str, ...] property readonly

Get a tuple of the names of the functions in the library.

Returns:

Type Description
Tuple[str, ...]

The names of the functions in the library.

name: str property readonly

The library name.

__init__(self, *functions, *, name='default_function_library_name') special

Parameters:

Name Type Description Default
name

obj:str, optional): The name of the function library.

'default_function_library_name'
*functions

obj:FunctionData, optional): A list of functions to initialize the object.

()
Source code in classiq/quantum_functions/function_library.py
def __init__(self, *functions, name: str = DEFAULT_FUNCTION_LIBRARY_NAME):
    """
    Args:
        name (:obj:`str`, optional): The name of the function library.
        *functions (:obj:`FunctionData`, optional): A list of functions to initialize the object.
    """
    self._data = FunctionLibraryData(name=name)
    self._params: Dict[str, CustomFunction] = dict()

    for f in functions:
        self.add_function(f)
add_function(self, function_data, override_existing_functions=False)

Adds a function to the function library.

Parameters:

Name Type Description Default
function_data FunctionData

The function data object.

required
override_existing_functions

obj:bool, optional): Defaults to False.

False

Returns:

Type Description
CustomFunction

The custom function parameters.

Source code in classiq/quantum_functions/function_library.py
def add_function(
    self,
    function_data: Union[FunctionData, QuantumFunction],
    override_existing_functions: bool = False,
) -> CustomFunction:
    """Adds a function to the function library.

    Args:
        function_data (FunctionData): The function data object.
        override_existing_functions (:obj:`bool`, optional): Defaults to False.

    Returns:
        The custom function parameters.
    """
    if isinstance(function_data, QuantumFunction):
        function_data = function_data.function_data

    if not isinstance(function_data, FunctionData):
        raise ClassiqValueError(
            f"FunctionData object expected, got {function_data.__class__.__name__}"
        )

    function_name = function_data.name
    if (
        not override_existing_functions
        and function_name in self._data.function_dict
    ):
        raise ClassiqValueError("Cannot override existing functions.")

    if function_data.function_type == FunctionType.CompositeFunction:
        for call in function_data.logic_flow:
            if not isinstance(call.function_params, CustomFunction):
                continue
            FunctionLibraryData.validate_function_in_library(
                library=self._data,
                function_params=call.function_params,
                error_handler=ClassiqValueError,
            )

    self._data.function_dict[function_name] = function_data
    self._params[function_name] = self._to_params(function_data)
    return self.get_function(function_name=function_name)
remove_function(self, function_name)

Removes a function from the function library.

Parameters:

Name Type Description Default
function_name str

The name of the function.

required

Returns:

Type Description
FunctionData

The removed function data.

Source code in classiq/quantum_functions/function_library.py
def remove_function(self, function_name: str) -> FunctionData:
    """Removes a function from the function library.

    Args:
        function_name (str): The name of the function.

    Returns:
        The removed function data.
    """
    self._params.pop(function_name)
    return self._data.function_dict.pop(function_name)

quantum_register

QReg

A Quantum Register - A logical collection of several qubits.

If the class has public attributes, they may be documented here in an Attributes section and follow the same formatting as a function's Args section. Alternatively, attributes may be documented inline with the attribute's declaration (see init method below).

Properties created with the @property decorator should be documented in the property's getter method.

Attributes:

Name Type Description
size int

The amount of qubits.

Source code in classiq/quantum_register.py
class QReg:
    """A Quantum Register - A logical collection of several qubits.

    If the class has public attributes, they may be documented here
    in an ``Attributes`` section and follow the same formatting as a
    function's ``Args`` section. Alternatively, attributes may be documented
    inline with the attribute's declaration (see __init__ method below).

    Properties created with the ``@property`` decorator should be documented
    in the property's getter method.

    Attributes:
        size (int): The amount of qubits.

    """

    is_signed: Optional[bool] = None
    fraction_places: Optional[int] = None

    # Object initialization
    def __init__(self, size: int) -> None:
        self.size: int = size

        self._wire: Wire = Wire()

        self._available_qubits: List[_Qubit] = [_Qubit() for _ in range(self.size)]

    def __class_getitem__(cls, params):
        # Supporting python 3.7+, thus returning `typing._GenericAlias` instead of `types.GenericAlias`
        if type(params) is int:
            return _GenericAliasWithSize(cls, params, inst=True, special=False)

        raise ClassiqQRegError(f"Invalid size: {params} ; int required")

    # Exported functions
    def __len__(self) -> int:
        return self.size

    # Exported functions - Availability
    @property
    def is_available(self) -> bool:
        """bool: Checks whether this QReg is available."""
        return all(self._available_qubits)

    @property
    def is_consumed(self) -> bool:
        """bool: Checks whether this QReg is consumed."""
        return not self.is_available

    def get_available_indexes(self) -> List[int]:
        """List[int]: Returns the indexes of the qubits that are available."""
        return [index for index, value in enumerate(self._available_qubits) if value]

    def consume(self) -> None:
        # This function is for internal usage.
        if self.is_consumed:
            raise ClassiqQRegError("Cannot consume a consumed QReg")

        for qubit in self._available_qubits:
            qubit.consume()

    # Wiring
    @property
    def wire(self) -> Wire:
        """Wire: Returns the wire associated with this QReg."""
        return self._wire

is_available: bool property readonly

bool: Checks whether this QReg is available.

is_consumed: bool property readonly

bool: Checks whether this QReg is consumed.

wire: Wire property readonly

Wire: Returns the wire associated with this QReg.

get_available_indexes(self)

List[int]: Returns the indexes of the qubits that are available.

Source code in classiq/quantum_register.py
def get_available_indexes(self) -> List[int]:
    """List[int]: Returns the indexes of the qubits that are available."""
    return [index for index, value in enumerate(self._available_qubits) if value]