Skip to content

Python SDK

Classiq SDK.

analyzer special

analyzer

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

Analyzer (AnalyzerUtilities)

Analyzer is the wrapper object for all analysis capabilities.

Source code in classiq/analyzer/analyzer.py
class Analyzer(AnalyzerUtilities, 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.
        """
        if circuit.qasm is None:
            raise ClassiqAnalyzerError(
                "Analysis requires a circuit with valid QASM code"
            )
        self._params: analysis_params.AnalysisParams = analysis_params.AnalysisParams(
            qasm=circuit.qasm
        )
        self.circuit: generator_result.GeneratedCircuit = circuit
        self.qc_graph: Optional[go.Figure] = None
        self.heatmap: Optional[go.Figure] = None
        self.gate_histogram: Optional[go.Figure] = None
        self.hardware_comparison_table: Optional[go.Figure] = None
        self.available_devices: ProviderAvailableDevices = dict()
        self.hardware_graphs: HardwareGraphs = dict()

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

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

        details = validate_type(
            obj=result,
            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.circuit)
        webbrowser.open_new_tab(urljoin(routes.ANALYZER_FULL_FE_APP, str(result.id)))

    async def get_available_devices_async(
        self, providers: Optional[List[ProviderNameEnum]] = None
    ) -> Dict[ProviderNameEnum, List[DeviceName]]:
        """Returns dict of the available devices by the providers. only devices
        with sufficient number of qubits are returns

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

        Returns:
            available devices (): dict of the available devices (Dict[str,List[str]]).
        """
        if providers is None:
            providers = list(AnalyzerProviderVendor)
        await self.request_available_devices_async(providers=providers)
        return {
            provider: self._filter_devices_by_qubits_count(provider)
            for provider in providers
        }

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

        Returns:
            None.
        """
        result = await ApiWrapper.call_qubits_connectivity_graphs_task(self._params)
        self.qc_graph = go.Figure(json.loads(result.details))

    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 self.qc_graph is None:
            await self.get_qubits_connectivity_async()
        self.qc_graph.show()  # type: ignore[union-attr]

    async def plot_hardware_connectivity_async(
        self,
        provider: Optional[ProviderNameEnum] = None,
        device: Optional[DeviceName] = None,
    ) -> VBox:
        """plot the hardware_connectivity graph. It is required to required  install the
        analyzer_sdk extra.

        Args:
            provider (): provider name (optional - string or `AnalyzerProviderVendor`).
            device (): device name (optional - string).
        Returns:
         hardware_connectivity_graph (): interactive graph.
        """

        self._validate_analyzer_extra()
        interactive_hardware = InteractiveHardware(
            circuit=self.circuit,
            params=self._params,
            available_devices=self.available_devices,
            hardware_graphs=self.hardware_graphs,
        )
        await interactive_hardware.enable_interactivity_async()
        if provider is not None:
            interactive_hardware.providers_combobox.value = provider
            if device is not None:
                interactive_hardware.devices_combobox.value = device

        return interactive_hardware.show_interactive_graph()

    async def get_hardware_comparison_table_async(
        self,
        providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
        devices: Optional[List[str]] = 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: providers (): List of providers (string or `AnalyzerProviderVendor`). if None, the table include all
        the available hardware.
        devices (): List of devices (string). if None, the table include all the available devices of the selected
        providers.
        Returns: None.
        """
        if providers is None:
            providers = list(AnalyzerProviderVendor)
        params = analysis_params.AnalysisHardwareListParams(
            qasm=self._params.qasm, providers=providers, devices=devices
        )
        result = await ApiWrapper.call_table_graphs_task(params=params)
        self.hardware_comparison_table = go.Figure(json.loads(result.details))

    async def plot_hardware_comparison_table_async(
        self,
        providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
        devices: Optional[List[str]] = None,
    ) -> None:
        """plot the comparison table. if it has not been created it, it first creates the table using all the
        available hardware.

        Returns:
            None.
        """
        await self._hardware_comparison_condition_async(
            providers=providers, devices=devices
        )
        self.hardware_comparison_table.show()  # type: ignore[union-attr]

    async def get_heatmap_async(self) -> None:
        """create a heatmap of the analysed circuit.

        Returns:
            None.
        """
        result = await ApiWrapper.call_heatmap_graphs(self._params)
        self.heatmap = _create_heatmap_graph(result, self.circuit.qubit_count)

    async def plot_heatmap_async(self) -> None:
        """plot the circuit heatmap. if it has not been created it, it will create the graph.

        Returns:
            None.
        """
        if self.heatmap is None:
            await self.get_heatmap_async()
        self.heatmap.show()  # type: ignore[union-attr]

    async def plot_gate_histogram_async(self) -> None:
        """plot the circuit gate histogram. if it has not been created it, it will create the graph.

        Returns:
            None.
        """
        if self.gate_histogram is None:
            await self.get_gate_histogram_async()
        self.gate_histogram.show()  # type: ignore[union-attr]

    async def get_gate_histogram_async(self) -> None:
        """create a gate histogram of the analysed circuit.

        Returns:
            None.
        """
        result = await ApiWrapper.call_gate_histogram_graphs(params=self._params)
        self.gate_histogram = _create_gate_histogram(
            result=result, num_qubits=self.circuit.qubit_count
        )

    async def hardware_aware_resynthesize_async(
        self, device: str, provider: Union[str, AnalyzerProviderVendor]
    ) -> generator_result.GeneratedCircuit:
        """resynthesize the analyzed circuit using its original model, and a new  backend preferences.

        Args:
            provider (): Provider company or cloud for the requested backend (string or `AnalyzerProviderVendor`).
            device (): Name of the requested backend"
        Returns:
            circuit (): resynthesize circuit (`GeneratedCircuit`).
        """

        update_preferences = self._validated_update_preferences(
            device=device, provider=provider
        )

        model_designer = ModelDesigner()
        model_designer._model = self.circuit.model.copy(deep=True)  # type: ignore[union-attr]
        return await model_designer.synthesize_async(preferences=update_preferences)

    async def optimized_hardware_resynthesize_async(
        self,
        comparison_property: Union[str, ComparisonProperties],
        providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
        devices: Optional[List[str]] = None,
    ) -> generator_result.GeneratedCircuit:
        """Re-synthesize the analyzed circuit using its original model, and a new backend preferences, which is the
         devices with the best fit to the selected comparison property.

        Args: comparison_property (): A comparison properties using to compare between the devices (string or
        `ComparisonProperties`).
        providers (): List of providers (string or `AnalyzerProviderVendor`). If None, the comparison include all the
        available hardware.
        devices (): List of devices (string). If None, the comparison include all the available devices of the selected
        providers.
        Returns: circuit (): resynthesize circuit (`GeneratedCircuit`).
        """
        optimized_device, optimized_provider = await self._get_optimized_hardware_async(
            providers=providers,
            devices=devices,
            comparison_property=comparison_property,
        )
        return await self.hardware_aware_resynthesize_async(
            provider=optimized_provider, device=optimized_device
        )

    async def _get_optimized_hardware_async(
        self,
        comparison_property: Union[str, ComparisonProperties],
        providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
        devices: Optional[List[str]] = None,
    ) -> Tuple[str, str]:
        await self._hardware_comparison_condition_async(
            providers=providers, devices=devices
        )
        optimized_device, optimized_provider = self._choose_optimized_hardware(
            comparison_property=comparison_property
        )
        return optimized_device, optimized_provider

    def _choose_optimized_hardware(
        self, comparison_property: Union[str, ComparisonProperties]
    ) -> Tuple[str, str]:
        comparison_params = AnalysisComparisonParams(property=comparison_property)
        if not isinstance(self.hardware_comparison_table, go.Figure):
            raise ClassiqAnalyzerError(
                "The analyzer does not contains a valid hardware comparison table"
            )
        column_names = self.hardware_comparison_table.data[0].header.values
        property_index = column_names.index(comparison_params.property.upper())

        sort_button = self.hardware_comparison_table.layout.updatemenus[0]
        sort_data = sort_button.buttons[property_index].args[0]["cells"]["values"]
        return sort_data[0][0], sort_data[1][0]

    def _validated_update_preferences(
        self, device: str, provider: Union[str, AnalyzerProviderVendor]
    ) -> Preferences:

        if not isinstance(self.circuit.model, Model):
            raise ClassiqAnalyzerError("The circuit does not contains a valid model")

        preferences_dict = self.circuit.model.preferences.dict()
        preferences_dict.update(
            dict(backend_service_provider=provider, backend_name=device)
        )

        return Preferences.parse_obj(preferences_dict)

    async def _hardware_comparison_condition_async(
        self,
        providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
        devices: Optional[List[str]] = None,
    ) -> None:
        if (
            providers is not None
            or devices is not None
            or self.hardware_comparison_table is None
        ):
            await self.get_hardware_comparison_table_async(
                providers=providers, devices=devices
            )

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

    @staticmethod
    def _validate_analyzer_extra() -> None:
        if find_ipywidgets is None:
            raise ClassiqAnalyzerError(
                "To use this method, please install the `analyzer sdk`. Run the  \
                following line: - pip install classiq[analyzer_sdk]"
            )
__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.
    """
    if circuit.qasm is None:
        raise ClassiqAnalyzerError(
            "Analysis requires a circuit with valid QASM code"
        )
    self._params: analysis_params.AnalysisParams = analysis_params.AnalysisParams(
        qasm=circuit.qasm
    )
    self.circuit: generator_result.GeneratedCircuit = circuit
    self.qc_graph: Optional[go.Figure] = None
    self.heatmap: Optional[go.Figure] = None
    self.gate_histogram: Optional[go.Figure] = None
    self.hardware_comparison_table: Optional[go.Figure] = None
    self.available_devices: ProviderAvailableDevices = dict()
    self.hardware_graphs: HardwareGraphs = dict()
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)

    details = validate_type(
        obj=result,
        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)

    details = validate_type(
        obj=result,
        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.circuit)
    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.circuit)
    webbrowser.open_new_tab(urljoin(routes.ANALYZER_FULL_FE_APP, str(result.id)))
get_available_devices(self, providers=None) async

Returns dict of the available devices by the providers. only devices with sufficient number of qubits are returns

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

Returns:

Type Description
available devices ()

dict of the available devices (Dict[str,List[str]]).

Source code in classiq/analyzer/analyzer.py
async def get_available_devices_async(
    self, providers: Optional[List[ProviderNameEnum]] = None
) -> Dict[ProviderNameEnum, List[DeviceName]]:
    """Returns dict of the available devices by the providers. only devices
    with sufficient number of qubits are returns

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

    Returns:
        available devices (): dict of the available devices (Dict[str,List[str]]).
    """
    if providers is None:
        providers = list(AnalyzerProviderVendor)
    await self.request_available_devices_async(providers=providers)
    return {
        provider: self._filter_devices_by_qubits_count(provider)
        for provider in providers
    }
get_available_devices_async(self, providers=None) async

Returns dict of the available devices by the providers. only devices with sufficient number of qubits are returns

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

Returns:

Type Description
available devices ()

dict of the available devices (Dict[str,List[str]]).

Source code in classiq/analyzer/analyzer.py
async def get_available_devices_async(
    self, providers: Optional[List[ProviderNameEnum]] = None
) -> Dict[ProviderNameEnum, List[DeviceName]]:
    """Returns dict of the available devices by the providers. only devices
    with sufficient number of qubits are returns

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

    Returns:
        available devices (): dict of the available devices (Dict[str,List[str]]).
    """
    if providers is None:
        providers = list(AnalyzerProviderVendor)
    await self.request_available_devices_async(providers=providers)
    return {
        provider: self._filter_devices_by_qubits_count(provider)
        for provider in providers
    }
get_gate_histogram(self) async

create a gate histogram of the analysed circuit.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def get_gate_histogram_async(self) -> None:
    """create a gate histogram of the analysed circuit.

    Returns:
        None.
    """
    result = await ApiWrapper.call_gate_histogram_graphs(params=self._params)
    self.gate_histogram = _create_gate_histogram(
        result=result, num_qubits=self.circuit.qubit_count
    )
get_gate_histogram_async(self) async

create a gate histogram of the analysed circuit.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def get_gate_histogram_async(self) -> None:
    """create a gate histogram of the analysed circuit.

    Returns:
        None.
    """
    result = await ApiWrapper.call_gate_histogram_graphs(params=self._params)
    self.gate_histogram = _create_gate_histogram(
        result=result, num_qubits=self.circuit.qubit_count
    )
get_hardware_comparison_table(self, providers=None, devices=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: providers (): List of providers (string or AnalyzerProviderVendor). if None, the table include all the available hardware. devices (): List of devices (string). if None, the table include all the available devices of the selected providers. Returns: None.

Source code in classiq/analyzer/analyzer.py
async def get_hardware_comparison_table_async(
    self,
    providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
    devices: Optional[List[str]] = 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: providers (): List of providers (string or `AnalyzerProviderVendor`). if None, the table include all
    the available hardware.
    devices (): List of devices (string). if None, the table include all the available devices of the selected
    providers.
    Returns: None.
    """
    if providers is None:
        providers = list(AnalyzerProviderVendor)
    params = analysis_params.AnalysisHardwareListParams(
        qasm=self._params.qasm, providers=providers, devices=devices
    )
    result = await ApiWrapper.call_table_graphs_task(params=params)
    self.hardware_comparison_table = go.Figure(json.loads(result.details))
get_hardware_comparison_table_async(self, providers=None, devices=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: providers (): List of providers (string or AnalyzerProviderVendor). if None, the table include all the available hardware. devices (): List of devices (string). if None, the table include all the available devices of the selected providers. Returns: None.

Source code in classiq/analyzer/analyzer.py
async def get_hardware_comparison_table_async(
    self,
    providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
    devices: Optional[List[str]] = 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: providers (): List of providers (string or `AnalyzerProviderVendor`). if None, the table include all
    the available hardware.
    devices (): List of devices (string). if None, the table include all the available devices of the selected
    providers.
    Returns: None.
    """
    if providers is None:
        providers = list(AnalyzerProviderVendor)
    params = analysis_params.AnalysisHardwareListParams(
        qasm=self._params.qasm, providers=providers, devices=devices
    )
    result = await ApiWrapper.call_table_graphs_task(params=params)
    self.hardware_comparison_table = go.Figure(json.loads(result.details))
get_heatmap(self) async

create a heatmap of the analysed circuit.

Returns:

Type Description
None

None.

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

    Returns:
        None.
    """
    result = await ApiWrapper.call_heatmap_graphs(self._params)
    self.heatmap = _create_heatmap_graph(result, self.circuit.qubit_count)
get_heatmap_async(self) async

create a heatmap of the analysed circuit.

Returns:

Type Description
None

None.

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

    Returns:
        None.
    """
    result = await ApiWrapper.call_heatmap_graphs(self._params)
    self.heatmap = _create_heatmap_graph(result, self.circuit.qubit_count)
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_qubits_connectivity_graphs_task(self._params)
    self.qc_graph = go.Figure(json.loads(result.details))
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_qubits_connectivity_graphs_task(self._params)
    self.qc_graph = go.Figure(json.loads(result.details))
hardware_aware_resynthesize(self, device, provider) async

resynthesize the analyzed circuit using its original model, and a new backend preferences.

Parameters:

Name Type Description Default
provider

Provider company or cloud for the requested backend (string or AnalyzerProviderVendor).

required
device

Name of the requested backend"

required

Returns:

Type Description
circuit ()

resynthesize circuit (GeneratedCircuit).

Source code in classiq/analyzer/analyzer.py
async def hardware_aware_resynthesize_async(
    self, device: str, provider: Union[str, AnalyzerProviderVendor]
) -> generator_result.GeneratedCircuit:
    """resynthesize the analyzed circuit using its original model, and a new  backend preferences.

    Args:
        provider (): Provider company or cloud for the requested backend (string or `AnalyzerProviderVendor`).
        device (): Name of the requested backend"
    Returns:
        circuit (): resynthesize circuit (`GeneratedCircuit`).
    """

    update_preferences = self._validated_update_preferences(
        device=device, provider=provider
    )

    model_designer = ModelDesigner()
    model_designer._model = self.circuit.model.copy(deep=True)  # type: ignore[union-attr]
    return await model_designer.synthesize_async(preferences=update_preferences)
hardware_aware_resynthesize_async(self, device, provider) async

resynthesize the analyzed circuit using its original model, and a new backend preferences.

Parameters:

Name Type Description Default
provider

Provider company or cloud for the requested backend (string or AnalyzerProviderVendor).

required
device

Name of the requested backend"

required

Returns:

Type Description
circuit ()

resynthesize circuit (GeneratedCircuit).

Source code in classiq/analyzer/analyzer.py
async def hardware_aware_resynthesize_async(
    self, device: str, provider: Union[str, AnalyzerProviderVendor]
) -> generator_result.GeneratedCircuit:
    """resynthesize the analyzed circuit using its original model, and a new  backend preferences.

    Args:
        provider (): Provider company or cloud for the requested backend (string or `AnalyzerProviderVendor`).
        device (): Name of the requested backend"
    Returns:
        circuit (): resynthesize circuit (`GeneratedCircuit`).
    """

    update_preferences = self._validated_update_preferences(
        device=device, provider=provider
    )

    model_designer = ModelDesigner()
    model_designer._model = self.circuit.model.copy(deep=True)  # type: ignore[union-attr]
    return await model_designer.synthesize_async(preferences=update_preferences)
optimized_hardware_resynthesize(self, comparison_property, providers=None, devices=None) async

Re-synthesize the analyzed circuit using its original model, and a new backend preferences, which is the devices with the best fit to the selected comparison property.

Args: comparison_property (): A comparison properties using to compare between the devices (string or ComparisonProperties). providers (): List of providers (string or AnalyzerProviderVendor). If None, the comparison include all the available hardware. devices (): List of devices (string). If None, the comparison include all the available devices of the selected providers. Returns: circuit (): resynthesize circuit (GeneratedCircuit).

Source code in classiq/analyzer/analyzer.py
async def optimized_hardware_resynthesize_async(
    self,
    comparison_property: Union[str, ComparisonProperties],
    providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
    devices: Optional[List[str]] = None,
) -> generator_result.GeneratedCircuit:
    """Re-synthesize the analyzed circuit using its original model, and a new backend preferences, which is the
     devices with the best fit to the selected comparison property.

    Args: comparison_property (): A comparison properties using to compare between the devices (string or
    `ComparisonProperties`).
    providers (): List of providers (string or `AnalyzerProviderVendor`). If None, the comparison include all the
    available hardware.
    devices (): List of devices (string). If None, the comparison include all the available devices of the selected
    providers.
    Returns: circuit (): resynthesize circuit (`GeneratedCircuit`).
    """
    optimized_device, optimized_provider = await self._get_optimized_hardware_async(
        providers=providers,
        devices=devices,
        comparison_property=comparison_property,
    )
    return await self.hardware_aware_resynthesize_async(
        provider=optimized_provider, device=optimized_device
    )
optimized_hardware_resynthesize_async(self, comparison_property, providers=None, devices=None) async

Re-synthesize the analyzed circuit using its original model, and a new backend preferences, which is the devices with the best fit to the selected comparison property.

Args: comparison_property (): A comparison properties using to compare between the devices (string or ComparisonProperties). providers (): List of providers (string or AnalyzerProviderVendor). If None, the comparison include all the available hardware. devices (): List of devices (string). If None, the comparison include all the available devices of the selected providers. Returns: circuit (): resynthesize circuit (GeneratedCircuit).

Source code in classiq/analyzer/analyzer.py
async def optimized_hardware_resynthesize_async(
    self,
    comparison_property: Union[str, ComparisonProperties],
    providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
    devices: Optional[List[str]] = None,
) -> generator_result.GeneratedCircuit:
    """Re-synthesize the analyzed circuit using its original model, and a new backend preferences, which is the
     devices with the best fit to the selected comparison property.

    Args: comparison_property (): A comparison properties using to compare between the devices (string or
    `ComparisonProperties`).
    providers (): List of providers (string or `AnalyzerProviderVendor`). If None, the comparison include all the
    available hardware.
    devices (): List of devices (string). If None, the comparison include all the available devices of the selected
    providers.
    Returns: circuit (): resynthesize circuit (`GeneratedCircuit`).
    """
    optimized_device, optimized_provider = await self._get_optimized_hardware_async(
        providers=providers,
        devices=devices,
        comparison_property=comparison_property,
    )
    return await self.hardware_aware_resynthesize_async(
        provider=optimized_provider, device=optimized_device
    )
plot_gate_histogram(self) async

plot the circuit gate histogram. if it has not been created it, it will create the graph.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_gate_histogram_async(self) -> None:
    """plot the circuit gate histogram. if it has not been created it, it will create the graph.

    Returns:
        None.
    """
    if self.gate_histogram is None:
        await self.get_gate_histogram_async()
    self.gate_histogram.show()  # type: ignore[union-attr]
plot_gate_histogram_async(self) async

plot the circuit gate histogram. if it has not been created it, it will create the graph.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_gate_histogram_async(self) -> None:
    """plot the circuit gate histogram. if it has not been created it, it will create the graph.

    Returns:
        None.
    """
    if self.gate_histogram is None:
        await self.get_gate_histogram_async()
    self.gate_histogram.show()  # type: ignore[union-attr]
plot_hardware_comparison_table(self, providers=None, devices=None) 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,
    providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
    devices: Optional[List[str]] = None,
) -> None:
    """plot the comparison table. if it has not been created it, it first creates the table using all the
    available hardware.

    Returns:
        None.
    """
    await self._hardware_comparison_condition_async(
        providers=providers, devices=devices
    )
    self.hardware_comparison_table.show()  # type: ignore[union-attr]
plot_hardware_comparison_table_async(self, providers=None, devices=None) 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,
    providers: Optional[List[Union[str, AnalyzerProviderVendor]]] = None,
    devices: Optional[List[str]] = None,
) -> None:
    """plot the comparison table. if it has not been created it, it first creates the table using all the
    available hardware.

    Returns:
        None.
    """
    await self._hardware_comparison_condition_async(
        providers=providers, devices=devices
    )
    self.hardware_comparison_table.show()  # type: ignore[union-attr]
plot_hardware_connectivity(self, provider=None, device=None) async

plot the hardware_connectivity graph. It is required to required install the analyzer_sdk extra.

Parameters:

Name Type Description Default
provider

provider name (optional - string or AnalyzerProviderVendor).

None
device

device name (optional - string).

None

Returns:

Type Description
hardware_connectivity_graph ()

interactive graph.

Source code in classiq/analyzer/analyzer.py
async def plot_hardware_connectivity_async(
    self,
    provider: Optional[ProviderNameEnum] = None,
    device: Optional[DeviceName] = None,
) -> VBox:
    """plot the hardware_connectivity graph. It is required to required  install the
    analyzer_sdk extra.

    Args:
        provider (): provider name (optional - string or `AnalyzerProviderVendor`).
        device (): device name (optional - string).
    Returns:
     hardware_connectivity_graph (): interactive graph.
    """

    self._validate_analyzer_extra()
    interactive_hardware = InteractiveHardware(
        circuit=self.circuit,
        params=self._params,
        available_devices=self.available_devices,
        hardware_graphs=self.hardware_graphs,
    )
    await interactive_hardware.enable_interactivity_async()
    if provider is not None:
        interactive_hardware.providers_combobox.value = provider
        if device is not None:
            interactive_hardware.devices_combobox.value = device

    return interactive_hardware.show_interactive_graph()
plot_hardware_connectivity_async(self, provider=None, device=None) async

plot the hardware_connectivity graph. It is required to required install the analyzer_sdk extra.

Parameters:

Name Type Description Default
provider

provider name (optional - string or AnalyzerProviderVendor).

None
device

device name (optional - string).

None

Returns:

Type Description
hardware_connectivity_graph ()

interactive graph.

Source code in classiq/analyzer/analyzer.py
async def plot_hardware_connectivity_async(
    self,
    provider: Optional[ProviderNameEnum] = None,
    device: Optional[DeviceName] = None,
) -> VBox:
    """plot the hardware_connectivity graph. It is required to required  install the
    analyzer_sdk extra.

    Args:
        provider (): provider name (optional - string or `AnalyzerProviderVendor`).
        device (): device name (optional - string).
    Returns:
     hardware_connectivity_graph (): interactive graph.
    """

    self._validate_analyzer_extra()
    interactive_hardware = InteractiveHardware(
        circuit=self.circuit,
        params=self._params,
        available_devices=self.available_devices,
        hardware_graphs=self.hardware_graphs,
    )
    await interactive_hardware.enable_interactivity_async()
    if provider is not None:
        interactive_hardware.providers_combobox.value = provider
        if device is not None:
            interactive_hardware.devices_combobox.value = device

    return interactive_hardware.show_interactive_graph()
plot_heatmap(self) async

plot the circuit heatmap. if it has not been created it, it will create the graph.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_heatmap_async(self) -> None:
    """plot the circuit heatmap. if it has not been created it, it will create the graph.

    Returns:
        None.
    """
    if self.heatmap is None:
        await self.get_heatmap_async()
    self.heatmap.show()  # type: ignore[union-attr]
plot_heatmap_async(self) async

plot the circuit heatmap. if it has not been created it, it will create the graph.

Returns:

Type Description
None

None.

Source code in classiq/analyzer/analyzer.py
async def plot_heatmap_async(self) -> None:
    """plot the circuit heatmap. if it has not been created it, it will create the graph.

    Returns:
        None.
    """
    if self.heatmap is None:
        await self.get_heatmap_async()
    self.heatmap.show()  # type: ignore[union-attr]
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 self.qc_graph is None:
        await self.get_qubits_connectivity_async()
    self.qc_graph.show()  # type: ignore[union-attr]
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 self.qc_graph is None:
        await self.get_qubits_connectivity_async()
    self.qc_graph.show()  # type: ignore[union-attr]

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

applications special

chemistry special

pyscf_hamiltonian

generate_hamiltonian_from_pyscf(molecule_problem)

Conversion method from MoleculeProblem to HamiltonianProblem based on PYSCF chemistry package. The method extracts the one-body and two-body electron integrals, transfers them from atomic orbitals to molecular orbitals, and then presents them using FermionOperators.

Parameters:

Name Type Description Default
molecule_problem MoleculeProblem

chemical information of the input molecule.

required

Returns:

Type Description
HamiltonianProblem

HamiltonianProblem

Source code in classiq/applications/chemistry/pyscf_hamiltonian.py
def generate_hamiltonian_from_pyscf(
    molecule_problem: MoleculeProblem,
) -> HamiltonianProblem:
    """
    Conversion method from MoleculeProblem to HamiltonianProblem based on
    PYSCF chemistry package. The method extracts the one-body and two-body
    electron integrals, transfers them from atomic orbitals to molecular orbitals,
    and then presents them using FermionOperators.

    Args:
        molecule_problem (MoleculeProblem): chemical information of the input molecule.

    Returns:
        HamiltonianProblem

    """
    molecule = _to_pyscf_molecule(molecule_problem)

    # running pyscf driver
    rhf = scf.RHF(molecule)
    rhf.verbose = 0  # avoid unnecessary printing
    rhf.kernel()

    # extracting chemical properties from pyscf
    hcore = rhf.get_hcore()
    int2e = molecule.intor("int2e", aosym=1)
    mo_coeff = rhf.mo_coeff

    # atomic orbitals to molecular orbitals
    mo_1e_einsum = _truncate(
        array=np.einsum("pq,pi,qj->ij", hcore, mo_coeff, mo_coeff), threshold=1e-12
    )

    mo_2e_einsum = _truncate(
        array=np.einsum(
            "pqrs,pi,qj,rk,sl->ijkl", int2e, *[mo_coeff] * 4, optimize=True
        ),
        threshold=1e-12,
    )

    fermion_op = _to_summed_fermion_op(mo_1e_einsum) + _to_summed_fermion_op(
        mo_2e_einsum
    )
    return HamiltonianProblem(
        mapping=molecule_problem.mapping,
        z2_symmetries=molecule_problem.z2_symmetries,
        hamiltonian=fermion_op,
        num_particles=_get_num_particles(molecule),
    )

qnn special

data_sets

all_bits_to_one(n)

Return an integer of length n bits, where all the bits are 1

Source code in classiq/applications/qnn/data_sets.py
def all_bits_to_one(n: int) -> int:
    """
    Return an integer of length `n` bits, where all the bits are `1`
    """
    return (2**n) - 1
all_bits_to_zero(n)

Return an integer of length n bits, where all the bits are 0

Source code in classiq/applications/qnn/data_sets.py
def all_bits_to_zero(n: int) -> int:
    """
    Return an integer of length `n` bits, where all the bits are `0`
    """
    return 0
state_to_label(pure_state)

input: a Tensor of binary numbers (0 or 1) - the return value of a measurement output: probability (from that measurement) of measuring 0 (in other words, |0> translates to 100% chance for measuring |0> ==> return value is 1.0 |1> translates to 0% chance for measuring |0> ==> return value is 0.0 )

Source code in classiq/applications/qnn/data_sets.py
def state_to_label(pure_state: Tensor) -> Tensor:
    """
    input: a `Tensor` of binary numbers (0 or 1) - the return value of a measurement
    output: probability (from that measurement) of measuring 0
    (in other words,
        |0> translates to 100% chance for measuring |0> ==> return value is 1.0
        |1> translates to   0% chance for measuring |0> ==> return value is 0.0
    )
    """
    # |0> means 100% chance to get |0> ==> 100% == 1.0
    # |1> means   0% chance to get |0> ==>   0% == 0.0

    # This line basically does `1 - bool(pure_state)`
    return 1 - pure_state.bool().int()
state_to_weights(pure_state)

input: a Tensor of binary numbers (0 or 1) output: the required angle of rotation for Rx (in other words, |0> translates to no rotation, and |1> translates to pi)

Source code in classiq/applications/qnn/data_sets.py
def state_to_weights(pure_state: Tensor) -> Tensor:
    """
    input: a `Tensor` of binary numbers (0 or 1)
    output: the required angle of rotation for `Rx`
    (in other words, |0> translates to no rotation, and |1> translates to `pi`)
    """
    # |0> requires a rotation by 0
    # |1> requires a rotation by pi
    return pure_state.bool().int() * np.pi

qlayer

QLayer (Module)
Source code in classiq/applications/qnn/qlayer.py
class QLayer(nn.Module):
    def __init__(
        self,
        circuit: Circuit,
        execute: ExecuteFunciton,
        post_process: PostProcessFunction,
        # Optional parameters:
        head_start: Union[float, Tensor, None] = None,
        # Experimental parameters:
        calc_num_out_features: CalcNumOutFeatures = calc_num_out_features_single_output,
    ):
        validate_circuit(circuit)

        super().__init__()

        self._execute = execute
        self._post_process = post_process
        self._head_start = head_start

        self.circuit = circuit

        weights, _ = extract_parameters(circuit)
        self.in_features = len(weights)
        self.out_features = calc_num_out_features(circuit)

        self._initialize_parameters()

    def _initialize_parameters(self) -> None:
        shape = (self.out_features, self.in_features)
        if self._head_start is None:
            value = torch.rand(shape)
        elif isinstance(self._head_start, float):
            value = torch.zeros(shape) + self._head_start
        else:
            raise ClassiqQNNError(
                f"Unsupported feature - head_start of type {type(self._head_start)}"
            )

        self.weight = Parameter(value)

    def forward(self, x: Tensor) -> Tensor:
        return QLayerFunction.apply(
            x, self.weight, self.circuit, self._execute, self._post_process
        )
forward(self, x)

Defines the computation performed at every call.

Should be overridden by all subclasses.

.. note:: Although the recipe for forward pass needs to be defined within this function, one should call the :class:Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

Source code in classiq/applications/qnn/qlayer.py
def forward(self, x: Tensor) -> Tensor:
    return QLayerFunction.apply(
        x, self.weight, self.circuit, self._execute, self._post_process
    )
QLayerFunction (Function)
Source code in classiq/applications/qnn/qlayer.py
class QLayerFunction(torch.autograd.Function):
    @staticmethod
    def forward(  # type: ignore[override]
        ctx,
        inputs: Tensor,
        weights: Tensor,
        circuit: Circuit,
        execute: ExecuteFunciton,
        post_process: PostProcessFunction,
    ) -> Tensor:
        """
        This function receives:
            inputs: a 2D Tensor of floats - (batch_size, in_features)
            weights: a 2D Tensor of floats - (out_features, num_weights)
            circuit: a `GeneratedCircuit` object
            execute: a function taking a `GeneratedCircuit` and `MultipleArguments`
                and returning `MultipleExecutionDetails`
            post_process: a function taking a single `ExecutionDetails`
                and returning a `Tensor`
        """
        validate_circuit(circuit)

        # save for backward
        ctx.save_for_backward(inputs, weights)
        ctx.circuit = circuit
        ctx.execute = execute
        ctx.post_process = post_process
        ctx.quantum_gradient = SimpleQuantumGradient(circuit, execute, post_process)

        ctx.batch_size, ctx.num_in_features = inputs.shape
        ctx.num_out_features, ctx.num_weights = weights.shape

        # Todo: avoid computing `_get_extracted_parameters` on every `forward`
        extracted_parameters = extract_parameters(circuit)

        # Todo: avoid defining `convert_tensors_to_arguments` on every `forward`
        def convert_tensors_to_arguments(
            inputs_: Tensor, weights_: Tensor
        ) -> MultipleArguments:
            arguments = map_parameters(
                extracted_parameters,
                inputs_,
                weights_,
            )
            return (arguments,)

        return iter_inputs_weights(
            inputs,
            weights,
            convert_tensors_to_arguments,
            functools.partial(execute, circuit),
            post_process,
            expected_shape=(ctx.batch_size, ctx.num_out_features),
        )

    @staticmethod
    def backward(  # type: ignore[override]
        ctx, grad_output: Tensor
    ) -> Tuple[Optional[Tensor], Optional[Tensor], None, None, None]:
        """
        grad_output: Tensor
            is of shape (ctx.batch_size, ctx.num_out_features)
        """
        inputs, weights = ctx.saved_tensors

        grad_weights = grad_inputs = None
        grad_circuit = grad_execute = grad_post_process = None

        if ctx.needs_input_grad[1]:
            grad_weights = ctx.quantum_gradient.gradient(inputs, weights)

            # For batch_size=8 ; num_out_features=1 ; num_weights=6
            # grad_output is 8x1
            # grad_weight is 8x1x6 (before the line below)
            #      weight is 1x6
            # result should be 1x6 (which is grad_weights after the line below)
            #   so grad_output.T * grad_weight
            grad_weights = torch.tensordot(
                grad_weights,
                grad_output,
                dims=[[0], [0]],
            )
            grad_weights = grad_weights.squeeze(-1)

        if ctx.needs_input_grad[0]:
            raise ClassiqTorchError("Stay tuned! coming next version")
        if any(ctx.needs_input_grad[i] for i in (2, 3, 4)):
            raise ClassiqTorchError(
                f"Grad required for unknown type: {ctx.needs_input_grad}"
            )

        return grad_inputs, grad_weights, grad_circuit, grad_execute, grad_post_process
backward(ctx, grad_output) staticmethod

Tensor

is of shape (ctx.batch_size, ctx.num_out_features)

Source code in classiq/applications/qnn/qlayer.py
@staticmethod
def backward(  # type: ignore[override]
    ctx, grad_output: Tensor
) -> Tuple[Optional[Tensor], Optional[Tensor], None, None, None]:
    """
    grad_output: Tensor
        is of shape (ctx.batch_size, ctx.num_out_features)
    """
    inputs, weights = ctx.saved_tensors

    grad_weights = grad_inputs = None
    grad_circuit = grad_execute = grad_post_process = None

    if ctx.needs_input_grad[1]:
        grad_weights = ctx.quantum_gradient.gradient(inputs, weights)

        # For batch_size=8 ; num_out_features=1 ; num_weights=6
        # grad_output is 8x1
        # grad_weight is 8x1x6 (before the line below)
        #      weight is 1x6
        # result should be 1x6 (which is grad_weights after the line below)
        #   so grad_output.T * grad_weight
        grad_weights = torch.tensordot(
            grad_weights,
            grad_output,
            dims=[[0], [0]],
        )
        grad_weights = grad_weights.squeeze(-1)

    if ctx.needs_input_grad[0]:
        raise ClassiqTorchError("Stay tuned! coming next version")
    if any(ctx.needs_input_grad[i] for i in (2, 3, 4)):
        raise ClassiqTorchError(
            f"Grad required for unknown type: {ctx.needs_input_grad}"
        )

    return grad_inputs, grad_weights, grad_circuit, grad_execute, grad_post_process
forward(ctx, inputs, weights, circuit, execute, post_process) staticmethod

This function receives: inputs: a 2D Tensor of floats - (batch_size, in_features) weights: a 2D Tensor of floats - (out_features, num_weights) circuit: a GeneratedCircuit object !!! execute "a function taking a GeneratedCircuit and MultipleArguments" and returning MultipleExecutionDetails !!! post_process "a function taking a single ExecutionDetails" and returning a Tensor

Source code in classiq/applications/qnn/qlayer.py
@staticmethod
def forward(  # type: ignore[override]
    ctx,
    inputs: Tensor,
    weights: Tensor,
    circuit: Circuit,
    execute: ExecuteFunciton,
    post_process: PostProcessFunction,
) -> Tensor:
    """
    This function receives:
        inputs: a 2D Tensor of floats - (batch_size, in_features)
        weights: a 2D Tensor of floats - (out_features, num_weights)
        circuit: a `GeneratedCircuit` object
        execute: a function taking a `GeneratedCircuit` and `MultipleArguments`
            and returning `MultipleExecutionDetails`
        post_process: a function taking a single `ExecutionDetails`
            and returning a `Tensor`
    """
    validate_circuit(circuit)

    # save for backward
    ctx.save_for_backward(inputs, weights)
    ctx.circuit = circuit
    ctx.execute = execute
    ctx.post_process = post_process
    ctx.quantum_gradient = SimpleQuantumGradient(circuit, execute, post_process)

    ctx.batch_size, ctx.num_in_features = inputs.shape
    ctx.num_out_features, ctx.num_weights = weights.shape

    # Todo: avoid computing `_get_extracted_parameters` on every `forward`
    extracted_parameters = extract_parameters(circuit)

    # Todo: avoid defining `convert_tensors_to_arguments` on every `forward`
    def convert_tensors_to_arguments(
        inputs_: Tensor, weights_: Tensor
    ) -> MultipleArguments:
        arguments = map_parameters(
            extracted_parameters,
            inputs_,
            weights_,
        )
        return (arguments,)

    return iter_inputs_weights(
        inputs,
        weights,
        convert_tensors_to_arguments,
        functools.partial(execute, circuit),
        post_process,
        expected_shape=(ctx.batch_size, ctx.num_out_features),
    )

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_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}

        request = execution_request.ExecutionRequest(
            preferences=self._preferences,
            execution_payload=execution_request.QuantumProgramExecution(**kwargs),
        )

        result = await ApiWrapper.call_execute_quantum_program(request=request)
        result.output_qubits_map = quantum_program.output_qubits_map
        return result

    async def execute_quantum_program_batch_async(
        self,
        quantum_program: QuantumProgram,
        arguments: MultipleArguments,
    ) -> MultipleExecutionDetails:
        kwargs = quantum_program.dict()
        kwargs["arguments"] = arguments
        if "output_qubits_map" in kwargs:
            del kwargs["output_qubits_map"]

        if "registers_initialization" in kwargs:
            del kwargs["registers_initialization"]

        request = execution_request.ExecutionRequest(
            preferences=self._preferences,
            execution_payload=execution_request.QuantumProgramBatchExecution(**kwargs),
        )

        result = await ApiWrapper.call_execute_quantum_program_batch(request=request)
        return result

    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.execution_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."
            )

        payload = execution_request.GenerationMetadataExecution(
            **generation_result.metadata.execution_metadata.dict()
        )
        request = execution_request.ExecutionRequest(
            preferences=self._preferences,
            execution_payload=payload,
        )

        if isinstance(payload.metadata, FinanceMetadata):
            return await ApiWrapper.call_execute_finance(request=request)
        elif isinstance(payload.metadata, GroverMetadata):
            return await ApiWrapper.call_execute_grover(request=request)
        raise ValueError("Couldn't find handler for execute task request.")

    async def execute_hamiltonian_minimization_async(
        self,
        hamiltonian_minimization_problem: HamiltonianMinimizationProblem,
    ) -> VQESolverResult:

        payload = execution_request.HamiltonianMinimizationProblemExecution(
            **hamiltonian_minimization_problem.dict()
        )

        request = execution_request.ExecutionRequest(
            preferences=self._preferences,
            execution_payload=payload,
        )

        return await ApiWrapper.call_execute_vqe(request=request)

    async def execute_async(
        self,
        arg: Union[GeneratedCircuit, QuantumProgram, str],
        arguments: Optional[Arguments] = None,
        initial_values: Optional[InitialConditions] = None,
    ):
        if isinstance(arg, GeneratedCircuit):
            if arg.metadata.execution_metadata is not None:
                return await self.execute_generated_circuit_async(generation_result=arg)
            else:
                program = arg.to_program(initial_values=initial_values)

        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(
            quantum_program=program, arguments=arguments
        )

__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)

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
        self._generated_inputs: Dict[IOName, Tuple[Wire, RegisterUserInput]] = dict()
        self._generated_outputs: Dict[IOName, Tuple[Wire, RegisterUserInput]] = dict()
        self._logic_flow_builder: LogicFlowBuilder = LogicFlowBuilder()

    def _verify_legal_wires(self, wires: WireOrWires) -> None:
        if isinstance(wires, Wire):
            wires = [wires]
        if any(
            wire not in self._generated_wires and not wire.wired_to_zero
            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)

    @staticmethod
    def _parse_control_states(
        control_states: Optional[Union[ControlState, Iterable[ControlState]]] = None
    ) -> List[ControlState]:
        if control_states is None:
            return list()
        elif isinstance(control_states, ControlState):
            return [control_states]
        return list(control_states)

    def _validate_unique_inputs(
        self, input_registers: Dict[IOName, RegisterUserInput]
    ) -> None:
        if any(name in self._generated_inputs for name in input_registers):
            raise ClassiqWiringError(_SAME_INPUT_NAME_ERROR_MSG)

    def create_inputs(
        self,
        input_registers: Dict[IOName, RegisterUserInput],
    ) -> Dict[IOName, LogicFlowInputWire]:
        wires_dict = {}
        self._validate_unique_inputs(input_registers)

        for name, reg in input_registers.items():
            wire = LogicFlowInputWire(input_name=name)
            self._generated_inputs[name] = (wire, reg)
            self._update_generated_wires(wire)
            wires_dict[name] = wire

        return wires_dict

    def set_outputs(
        self, outputs: Dict[IOName, Tuple[Wire, RegisterUserInput]]
    ) -> None:
        self._verify_legal_wires(wires=[output[0] for output in outputs.values()])
        for output_name, output in outputs.items():
            output_wire, output_register = output
            wire_name = output_wire.wire_name
            if isinstance(output_wire, LogicFlowInputWire):
                raise ClassiqWiringError(f"{_INPUT_AS_OUTPUT_ERROR_MSG} {output_name}")
            output_wire.set_as_output(output_wire_name=wire_name)
            self._generated_outputs[output_name] = (output_wire, output_register)

    def apply(
        self,
        function_name: Union[
            str,
            FunctionData,
            QuantumFunction,
        ],
        in_wires: Optional[SupportedInputArgs] = None,
        out_wires: Optional[SupportedInputArgs] = None,
        is_inverse: bool = False,
        release_by_inverse: bool = False,
        control_states: Optional[Union[ControlState, Iterable[ControlState]]] = None,
        should_control: bool = True,
        power: int = 1,
        call_name: Optional[str] = None,
    ) -> Dict[str, QregOrWire]:
        # if there's no function library, create one
        if self._function_library is None:
            self.include_library(FunctionLibrary())

        if isinstance(function_name, FunctionData):
            function_data = function_name
        elif isinstance(function_name, QuantumFunction):
            function_data = function_name.function_data
        else:
            function_data = None

        if function_data:
            if function_data not in self._function_library.data:  # type: ignore[union-attr]
                self._function_library.add_function(function_data)  # type: ignore[union-attr]

            function_name = function_data.name

        return self._add_function_call(
            CustomFunction.__name__,
            self._function_library.get_function(cast(str, function_name)),  # type: ignore[union-attr]
            in_wires=in_wires,
            out_wires=out_wires,
            is_inverse=is_inverse,
            release_by_inverse=release_by_inverse,
            control_states=control_states,
            should_control=should_control,
            power=power,
            call_name=call_name,
        )

    def release_qregs(self, qregs: Union[QReg, Collection[QReg]]) -> None:
        if isinstance(qregs, QReg):
            qregs = [qregs]
        for qreg in qregs:
            self._logic_flow_builder.connect_qreg_to_zero(qreg)

    def _add_function_call(
        self,
        function: str,
        params: function_params.FunctionParams,
        control_states: Optional[Union[ControlState, Iterable[ControlState]]] = None,
        in_wires: Optional[SupportedInputArgs] = None,
        out_wires: Optional[SupportedInputArgs] = None,
        is_inverse: bool = False,
        release_by_inverse: bool = False,
        should_control: bool = True,
        power: int = 1,
        call_name: Optional[str] = None,
    ) -> Dict[str, QregOrWire]:
        if function != type(params).__name__:
            raise ClassiqValueError(
                "The FunctionParams type does not match function name"
            )

        if (
            isinstance(params, CustomFunction)
            and self._function_library
            and params not in self._function_library.data
        ):
            raise ClassiqValueError("The function is not found in included library.")

        call = function_call.FunctionCall(
            function=function,
            function_params=params,
            is_inverse=is_inverse ^ (power < 0),
            release_by_inverse=release_by_inverse,
            control_states=self._parse_control_states(control_states),
            should_control=should_control,
            power=abs(power),
            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: SupportedInputArgs,
    ) -> None:
        if isinstance(in_wires, dict):
            self._connect_named_in_wires(call=call, in_wires=in_wires)
        elif isinstance(in_wires, QReg):
            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[QReg],
    ) -> None:
        call_inputs = call.function_params.get_io_names(
            function_params.IO.Input, call.is_inverse, call.control_states
        )

        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'
            )
        self._connect_named_in_wires(call, dict(zip(call_inputs, quantum_registers)))

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

        for input_name, in_wire in in_wires.items():
            if isinstance(in_wire, QReg):
                pin_name, pin_indices = self._get_pin_name_and_indices(
                    input_name, call.input_regs_dict
                )
                self._logic_flow_builder.connect_qreg_to_func_call(
                    in_wire, pin_name, call, pin_indices
                )
            else:
                in_wire.connect_wire_end(end_call=call, input_name=input_name)

    @staticmethod
    def _get_pin_name_and_indices(
        input_name: str, inputs_info: Dict[str, RegisterUserInput]
    ) -> Tuple[str, range]:
        name, slicing = function_call.FunctionCall.parse_io_slicing(input_name)
        pin_info = inputs_info.get(name)
        if pin_info is None:
            raise ClassiqWiringError(
                f"No register size information on input pin {name}"
            )
        indices = range(pin_info.size)[slicing]
        return name, indices

    def _connect_out_wires(
        self,
        call: function_call.FunctionCall,
        out_wires: SupportedInputArgs,
    ) -> Dict[str, QregOrWire]:
        if isinstance(out_wires, dict):
            wire_dict = self._connect_named_out_wires(call=call, out_wires=out_wires)
        elif isinstance(out_wires, QReg):
            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, QReg) for i in out_wires):
                raise ClassiqWiringError(
                    "When supplying an 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 for wire in wire_dict.values() if isinstance(wire, Wire)]
        )
        return wire_dict

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

        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, QReg):
                    self._logic_flow_builder.connect_func_call_to_qreg(
                        call, specified_output, out_wire
                    )
                else:
                    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[QReg],
    ) -> Dict[str, QregOrWire]:
        output_names = call.function_params.get_io_names(
            function_params.IO.Output, call.is_inverse, call.control_states
        )
        return self._connect_named_out_wires(
            call, dict(zip(output_names, quantum_registers))
        )

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

        if is_builtin_function_name:
            return functools.partial(self._add_function_call, 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)

        if (
            self._function_library is not None
            and item in self._function_library.function_factory_names
        ):
            return functools.partial(
                self._function_library.get_function_factory(item),
                add_method=functools.partial(
                    self._function_library.add_function,
                    override_existing_functions=True,
                ),
                apply_method=self.apply,
            )

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

    def __dir__(self):
        builtin_func_name = [
            func.__name__
            for func in function_param_list.function_param_library.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)

    @classmethod
    def from_model(cls, model: Model) -> ModelDesigner:
        return cls(**dict(model))

    @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

    def create_inputs(
        self, input_registers: Dict[IOName, RegisterUserInput]
    ) -> Dict[IOName, LogicFlowInputWire]:
        wires = super().create_inputs(input_registers=input_registers)
        self._model.inputs = self._generated_io_to_wire_name_dict(
            self._generated_inputs
        )
        return wires

    def set_outputs(
        self, outputs: Dict[IOName, Tuple[wire.Wire, RegisterUserInput]]
    ) -> None:
        super().set_outputs(outputs=outputs)
        self._model.outputs = self._generated_io_to_wire_name_dict(
            self._generated_outputs
        )

    @staticmethod
    def _generated_io_to_wire_name_dict(
        wire_obj_dict: Dict[IOName, Tuple[wire.Wire, RegisterUserInput]]
    ) -> Dict[IOName, WireName]:
        return {
            name: inner_wire.wire_name
            for name, (inner_wire, _) in wire_obj_dict.items()
        }

    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
        generation_result = await ApiWrapper.call_generation_task(self._model)
        generation_result.model = self._model.copy(deep=True)
        return generation_result

    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
    generation_result = await ApiWrapper.call_generation_task(self._model)
    generation_result.model = self._model.copy(deep=True)
    return generation_result
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
    generation_result = await ApiWrapper.call_generation_task(self._model)
    generation_result.model = self._model.copy(deep=True)
    return generation_result

quantum_functions special

annotation_parser

get_annotation_role(annotation, default_role)

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()
        self._func_factories: Dict[str, Type[QuantumFunctionFactory]] = dict()

        for f in functions:
            self.add_function(f)

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

    def get_function_factory(
        self, function_factory_name: str
    ) -> Type[QuantumFunctionFactory]:
        return self._func_factories[function_factory_name]

    def __getitem__(self, key: Any) -> CustomFunction:
        if isinstance(key, str):
            return self.get_function(key)
        else:
            raise ClassiqValueError("Invalid key")

    def add_function(
        self,
        function_data: Union[
            FunctionData, QuantumFunction, Type[QuantumFunctionFactory]
        ],
        override_existing_functions: bool = False,
    ) -> None:
        """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:
            None
        """
        if isinstance(function_data, type) and issubclass(
            function_data, QuantumFunctionFactory
        ):
            self._func_factories[function_data.__name__] = function_data
            return
        if isinstance(function_data, QuantumFunction):
            function_data = function_data.function_data

        if not isinstance(
            function_data, (ElementaryFunctionData, CompositeFunctionData)
        ):
            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 isinstance(function_data, CompositeFunctionData):
            for call in filter(
                lambda i: isinstance(i.function_params, CustomFunction),
                function_data.logic_flow,
            ):
                if self._data and call.function_params not in self.data:
                    raise ClassiqValueError(
                        "The function is not found in included library."
                    )

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

    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 function_factory_names(self) -> Tuple[str, ...]:
        return tuple(self._func_factories.keys())

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

    @staticmethod
    def _to_params(data: FunctionData) -> CustomFunction:
        params = CustomFunction(name=data.name)
        params.generate_ios(
            inputs=data.inputs,
            outputs=data.outputs,
        )
        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()
    self._func_factories: Dict[str, Type[QuantumFunctionFactory]] = 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
None

None

Source code in classiq/quantum_functions/function_library.py
def add_function(
    self,
    function_data: Union[
        FunctionData, QuantumFunction, Type[QuantumFunctionFactory]
    ],
    override_existing_functions: bool = False,
) -> None:
    """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:
        None
    """
    if isinstance(function_data, type) and issubclass(
        function_data, QuantumFunctionFactory
    ):
        self._func_factories[function_data.__name__] = function_data
        return
    if isinstance(function_data, QuantumFunction):
        function_data = function_data.function_data

    if not isinstance(
        function_data, (ElementaryFunctionData, CompositeFunctionData)
    ):
        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 isinstance(function_data, CompositeFunctionData):
        for call in filter(
            lambda i: isinstance(i.function_params, CustomFunction),
            function_data.logic_flow,
        ):
            if self._data and call.function_params not in self.data:
                raise ClassiqValueError(
                    "The function is not found in included library."
                )

    self._data.function_dict[function_name] = function_data
    self._params[function_name] = self._to_params(function_data)
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_function

QuantumFunctionFactory (ABC)

Provides the capability of creating parametrized user-defined functions.

Source code in classiq/quantum_functions/quantum_function.py
class QuantumFunctionFactory(ABC):
    """
    Provides the capability of creating parametrized user-defined functions.
    """

    def __init__(self, add_method: Callable, apply_method: Callable):
        self._apply_method = apply_method
        try:
            definition = self.definition
        except AttributeError as e:
            raise QuantumFunctionFactoryBadUsageError(
                f"{self.__class__.__name__} instance definition parsing failed."
            ) from e
        definition.function_data.name = str(self)
        add_method(definition)

    def __str__(self):
        str_list = [self.__class__.__name__.lower()]
        str_list.extend(
            f"{k}_{v}" for k, v in self.__dict__.items() if k != "_apply_method"
        )
        return "_".join(str_list)

    def __call__(self, *args, **kwargs):
        try:
            return self._apply_method(str(self), *args, **kwargs)
        except AttributeError as e:
            raise QuantumFunctionFactoryBadUsageError(
                f"Could not call {self.__class__.__name__}."
            ) from e

    @property
    @abstractmethod
    def definition(self) -> QuantumFunction:
        """
        Abstract method for providing the definition of the user function.
        The QuantumFunction object may be generated either directly, or using existing
        helper tools such as the @qfunc decorator.
        Instance attributes of the QuantumFunctionFactory may be used as parameters for the
        definition.

        Returns:
            The user-defined QuantumFunction object.
        """
definition: QuantumFunction property readonly

Abstract method for providing the definition of the user function. The QuantumFunction object may be generated either directly, or using existing helper tools such as the @qfunc decorator. Instance attributes of the QuantumFunctionFactory may be used as parameters for the definition.

Returns:

Type Description
QuantumFunction

The user-defined QuantumFunction object.

quantum_register

QReg

Represents a logical sequence of qubits. The QReg can be used as an in_wires or out_wires argument to ModelDesigner function calls, assisting in model connectivity.

Source code in classiq/quantum_register.py
class QReg:
    """Represents a logical sequence of qubits.
    The QReg can be used as an `in_wires` or `out_wires` argument to ModelDesigner function calls,
    assisting in model connectivity.
    """

    def __init__(self, size: int) -> None:
        """Initializes a new QReg with the specified number of qubits.

        Args:
            size (int): The number of qubits in the QReg.
        """
        if size <= 0:
            raise ClassiqQRegError(f"Cannot create {size} new qubits")
        self._qubits = [Qubit() for _ in range(size)]

    @classmethod
    def _from_qubits(cls, qubits: List[Qubit]) -> "QReg":
        if (
            not isinstance(qubits, list)
            or not all(isinstance(qubit, Qubit) for qubit in qubits)
            or len(qubits) == 0
        ):
            raise ClassiqQRegError(f"Cannot create QReg from {qubits}")
        qreg = cls(size=1)
        qreg._qubits = qubits
        return qreg

    def __getitem__(self, key: Union[int, slice]) -> "QReg":
        state = self._qubits[key]
        return QReg._from_qubits(state if isinstance(state, list) else [state])

    def __setitem__(self, key: Union[int, slice], value: "QReg") -> None:
        if isinstance(key, int) and len(value) != 1:
            raise ClassiqQRegError(
                f"Size mismatch: value size {len(value)}, expected size 1"
            )
        self._qubits[key] = value._qubits[0] if isinstance(key, int) else value._qubits  # type: ignore[call-overload]

    def __eq__(self, other) -> bool:
        return isinstance(other, QReg) and self._qubits == other._qubits

    @classmethod
    def concat(cls, left: "QReg", right: "QReg") -> "QReg":
        """Concatenate two QRegs.

        Args:
            left: The LHS QReg of the concatenation.
            right: The RHS QReg of the concatenation.

        Returns:
            A QReg representing the concatenation of the two given QRegs.

        """
        return cls._from_qubits(left._qubits + right._qubits)

    def __len__(self) -> int:
        return len(self._qubits)

    @property
    def qubits(self) -> List[Qubit]:
        return self._qubits

    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")

__init__(self, size) special

Initializes a new QReg with the specified number of qubits.

Parameters:

Name Type Description Default
size int

The number of qubits in the QReg.

required
Source code in classiq/quantum_register.py
def __init__(self, size: int) -> None:
    """Initializes a new QReg with the specified number of qubits.

    Args:
        size (int): The number of qubits in the QReg.
    """
    if size <= 0:
        raise ClassiqQRegError(f"Cannot create {size} new qubits")
    self._qubits = [Qubit() for _ in range(size)]

concat(left, right) classmethod

Concatenate two QRegs.

Parameters:

Name Type Description Default
left QReg

The LHS QReg of the concatenation.

required
right QReg

The RHS QReg of the concatenation.

required

Returns:

Type Description
QReg

A QReg representing the concatenation of the two given QRegs.

Source code in classiq/quantum_register.py
@classmethod
def concat(cls, left: "QReg", right: "QReg") -> "QReg":
    """Concatenate two QRegs.

    Args:
        left: The LHS QReg of the concatenation.
        right: The RHS QReg of the concatenation.

    Returns:
        A QReg representing the concatenation of the two given QRegs.

    """
    return cls._from_qubits(left._qubits + right._qubits)

RegisterRole (str, Enum)

An enumeration.

Source code in classiq/quantum_register.py
class RegisterRole(str, enum.Enum):
    INPUT = "input"
    OUTPUT = "output"
    AUXILIARY = "auxiliary"
    ZERO = "zero"