Lecture du code source Qiskit ~ Terra: lecture du backend Get, Call, Get Result

Que faites-vous?

Je crée Blueqat, une bibliothèque de calcul quantique, et je prévois de lire le code source de Qiskit, une bibliothèque de calcul quantique.

Dans la continuité de Dernière fois, cette fois, nous allons lire le processus de lecture de l'acquisition du backend, de l'appel et de l'obtention des résultats. Au lieu de lire la source du simulateur lui-même, nous lirons les interfaces autour du backend.

Présentation de Qiskit

Qiskit est une bibliothèque informatique quantique open source développée par IBM.

Qiskit est divisé en paquets comme indiqué ci-dessous, mais lors de l'installation, il est moins gênant de les installer avec pip install qiskit que de les faire séparément.

paquet rôle
Qiskit Terra Ceci est le package principal. Il comprend une classe pour créer un circuit, une fonction pour transpiler le circuit pour la machine réelle, une fonction pour atteindre l'API et la lancer sur la machine réelle, etc.
Qiskit Aer Comprend un simulateur de circuit quantique, généralement appelé depuis Qiskit Terra
Qiskit Ignis Ceci est une bibliothèque pour ceux qui veulent lutter contre le bruit lors de l'exécution d'un circuit quantique sur une machine réelle. Je n'ai jamais utilisé
Qiskit Aqua Une bibliothèque qui facilite l'utilisation des algorithmes quantiques

Cette fois, je lis une partie de Qiskit Terra.

https://github.com/Qiskit/qiskit-terra

Plus précisément, le code écrit en README.md

from qiskit import *
qc = QuantumCircuit(2, 2)
qc.h(0)
qc.cx(0, 1)
qc.measure([0,1], [0,1])
backend_sim = BasicAer.get_backend('qasm_simulator')
result = execute(qc, backend_sim).result()
print(result.get_counts(qc))

Parmi ceux-ci, lisez le flux après backend_sim = BasicAer.get_backend ('qasm_simulator'). Cependant, je ne lirai pas les détails du contenu du simulateur ou du transpile.

Continuez à lire la branche principale sur GitHub. L'ID de validation pour le moment est e7be587, mais il est mis à jour assez souvent et peut changer à la fin de l'article. Notez s'il vous plaît.

Lisez la ligne BasicAer.get_backend

backend_sim = BasicAer.get_backend('qasm_simulator')

Lisez cette ligne. BasicAer est dans [qiskit / fournisseurs / basicaer / init.py] de terra (https://github.com/Qiskit/qiskit-terra/blob/master/qiskit/providers/basicaer/init.py)

BasicAer = BasicAerProvider()

Est défini comme.

Lire BasicAerProvider

Lisez le BasicAerProvider dans qiskit / fournisseurs / basicaer / basicaerprovider.py.

SIMULATORS = [
    QasmSimulatorPy,
    StatevectorSimulatorPy,
    UnitarySimulatorPy
]


class BasicAerProvider(BaseProvider):
    """Provider for Basic Aer backends."""

    def __init__(self, *args, **kwargs):
        super().__init__(args, kwargs)

        # Populate the list of Basic Aer backends.
        self._backends = self._verify_backends()

    #Abréviation

    def _verify_backends(self):
        """
        Return the Basic Aer backends in `BACKENDS` that are
        effectively available (as some of them might depend on the presence
        of an optional dependency or on the existence of a binary).
        Returns:
            dict[str:BaseBackend]: a dict of Basic Aer backend instances for
                the backends that could be instantiated, keyed by backend name.
        """
        ret = OrderedDict()
        for backend_cls in SIMULATORS:
            try:
                backend_instance = self._get_backend_instance(backend_cls)
                backend_name = backend_instance.name()
                ret[backend_name] = backend_instance
            except QiskitError as err:
                # Ignore backends that could not be initialized.
                logger.info('Basic Aer backend %s is not available: %s',
                            backend_cls, str(err))
        return ret

Pour le moment, lisez le "init" de la classe parente "BaseProvider".

class BaseProvider(ABC):
    """Base class for a Backend Provider."""

    def __init__(self, *args, **kwargs):
        pass

Je ne fais rien.

Lisez BasicAerProvider._verify_backends

continuer

self._backends = self._verify_backends()

Lisez _verify_backends dans. J'ai collé le code ci-dessus, mais il est vraiment mauvais de faire autre chose que vérifier avec le nom verify ... Mettez-le de côté. Nous pouvons voir la structure selon laquelle «BasicAer» est un fournisseur et un fournisseur a un «Backend».

Il semble que les classes [QasmSimulatorPy, StatevectorSimulatorPy, UnitarySimulatorPy] définies dans SIMULATORS sont chacune _get_backend_instance et emballées dans un dictionnaire (ordonné) et renvoyées.

Lisons _get_backend_instance. La docstring est supprimée et citée.

    def _get_backend_instance(self, backend_cls):
        # Verify that the backend can be instantiated.
        try:
            backend_instance = backend_cls(provider=self)
        except Exception as err:
            raise QiskitError('Backend %s could not be instantiated: %s' %
                              (backend_cls, err))

        return backend_instance

Je crée juste une instance de la classe backend, mais je passe le fournisseur lui-même à l'instance. Le fournisseur et le backend sont dans une relation de référence mutuelle.

Lisez QasmSimulatorPy .__ init__

Ces classes backend elles-mêmes qiskit / fournisseurs / basicaer / qasm_simulator.py, qiskit / fournisseurs / basicaer / statevector_simulator. py, [qiskit / fournisseurs / basicaer / unitary_simulator.py](https: // github. Il est défini dans com / Qiskit / qiskit-terra / blob / master / qiskit / fournisseurs / basicaer / unitary_simulator.py). Regardons simplement QasmSimulatorPy .__ init__.

class QasmSimulatorPy(BaseBackend):
    """Python implementation of a qasm simulator."""

    MAX_QUBITS_MEMORY = int(log2(local_hardware_info()['memory'] * (1024 ** 3) / 16))

    DEFAULT_CONFIGURATION = {
        'backend_name': 'qasm_simulator',
        'backend_version': '2.0.0',
        'n_qubits': min(24, MAX_QUBITS_MEMORY),
        'url': 'https://github.com/Qiskit/qiskit-terra',
        'simulator': True,
        'local': True,
        'conditional': True,
        'open_pulse': False,
        'memory': True,
        'max_shots': 65536,
        'coupling_map': None,
        'description': 'A python simulator for qasm experiments',
        'basis_gates': ['u1', 'u2', 'u3', 'cx', 'id', 'unitary'],
        'gates': [
            {
                'name': 'u1',
                'parameters': ['lambda'],
                'qasm_def': 'gate u1(lambda) q { U(0,0,lambda) q; }'
            },
            {
                'name': 'u2',
                'parameters': ['phi', 'lambda'],
                'qasm_def': 'gate u2(phi,lambda) q { U(pi/2,phi,lambda) q; }'
            },
            {
                'name': 'u3',
                'parameters': ['theta', 'phi', 'lambda'],
                'qasm_def': 'gate u3(theta,phi,lambda) q { U(theta,phi,lambda) q; }'
            },
            {
                'name': 'cx',
                'parameters': ['c', 't'],
                'qasm_def': 'gate cx c,t { CX c,t; }'
            },
            {
                'name': 'id',
                'parameters': ['a'],
                'qasm_def': 'gate id a { U(0,0,0) a; }'
            },
            {
                'name': 'unitary',
                'parameters': ['matrix'],
                'qasm_def': 'unitary(matrix) q1, q2,...'
            }
        ]
    }

    DEFAULT_OPTIONS = {
        "initial_statevector": None,
        "chop_threshold": 1e-15
    }

    # Class level variable to return the final state at the end of simulation
    # This should be set to True for the statevector simulator
    SHOW_FINAL_STATE = False

    def __init__(self, configuration=None, provider=None):
        super().__init__(configuration=(
            configuration or QasmBackendConfiguration.from_dict(self.DEFAULT_CONFIGURATION)),
                         provider=provider)

        # Define attributes in __init__.
        self._local_random = np.random.RandomState()
        self._classical_memory = 0
        self._classical_register = 0
        self._statevector = 0
        self._number_of_cmembits = 0
        self._number_of_qubits = 0
        self._shots = 0
        self._memory = False
        self._initial_statevector = self.DEFAULT_OPTIONS["initial_statevector"]
        self._chop_threshold = self.DEFAULT_OPTIONS["chop_threshold"]
        self._qobj_config = None
        # TEMP
        self._sample_measure = False

J'ai l'impression d'avoir mis dans divers contextes. Ce que vous faites, c'est simplement initialiser des variables, et il ne semble pas que vous en fassiez beaucoup. Vous ne pouvez simuler que jusqu'à 24 bits quantiques. Je ne savais pas. (J'ai l'impression qu'il est assez peu réglé. Vous devriez pouvoir faire de votre mieux même avec un ordinateur normal)

Il semble que le backend lui-même ait un état qui semble être en cours, comme le nombre de plans et la mémoire. (Blueqat ose éviter une telle implémentation) L'idée ici peut devenir un peu plus claire à mesure que vous lisez le code.

Qu'est-ce que Basic Aer

--BasicAer est une instance de BasicAerProvider

Je comprends. Ensuite, jetons un œil à BasicAer.get_backend.

Lisez BasicAer.get_backend


    def get_backend(self, name=None, **kwargs):
        backends = self._backends.values()

        # Special handling of the `name` parameter, to support alias resolution
        # and deprecated names.
        if name:
            try:
                resolved_name = resolve_backend_name(
                    name, backends,
                    self._deprecated_backend_names(),
                    {}
                )
                name = resolved_name
            except LookupError:
                raise QiskitBackendNotFoundError(
                    "The '{}' backend is not installed in your system.".format(name))

        return super().get_backend(name=name, **kwargs)

Même si j'ai self._backends dans le dictionnaire, je suis curieux de le faire sans consulter le dictionnaire. Lisons resolution_backend_name.

Lire resolution_backend_name

resolution_backend_name est défini dans qiskit / provider / providersutils.py.

def resolve_backend_name(name, backends, deprecated, aliased):
    """Resolve backend name from a deprecated name or an alias.
    A group will be resolved in order of member priorities, depending on
    availability.
    Args:
        name (str): name of backend to resolve
        backends (list[BaseBackend]): list of available backends.
        deprecated (dict[str: str]): dict of deprecated names.
        aliased (dict[str: list[str]]): dict of aliased names.
    Returns:
        str: resolved name (name of an available backend)
    Raises:
        LookupError: if name cannot be resolved through regular available
            names, nor deprecated, nor alias names.
    """
    available = [backend.name() for backend in backends]

    resolved_name = deprecated.get(name, aliased.get(name, name))
    if isinstance(resolved_name, list):
        resolved_name = next((b for b in resolved_name if b in available), "")

    if resolved_name not in available:
        raise LookupError("backend '{}' not found.".format(name))

    if name in deprecated:
        logger.warning("Backend '%s' is deprecated. Use '%s'.", name,
                       resolved_name)

    return resolved_name

Avec ʻavailable`, faites une liste de noms de backend.

deprecated peut être vu par ceux qui utilisent Qiskit depuis un certain temps en regardant BasicAerProvider._deprecated_backend_names ().

    @staticmethod
    def _deprecated_backend_names():
        """Returns deprecated backend names."""
        return {
            'qasm_simulator_py': 'qasm_simulator',
            'statevector_simulator_py': 'statevector_simulator',
            'unitary_simulator_py': 'unitary_simulator',
            'local_qasm_simulator_py': 'qasm_simulator',
            'local_statevector_simulator_py': 'statevector_simulator',
            'local_unitary_simulator_py': 'unitary_simulator',
            'local_unitary_simulator': 'unitary_simulator',
            }

C'est un dictionnaire de {'old backend name': 'active backend name'}. Si le nom du backend que vous recherchez est dans obsolète, convertissez-le en nom de backend actif.

Pour «aliasé», j'ai recherché «qiskit-terra» et «qiskit-aer» mais je n'ai rien trouvé qui n'était pas vide, mais le nom du backend que je cherchais était dans le dictionnaire «aliasé». Si tel est le cas, il semble récupérer la liste des alias. Puisqu'il y a plusieurs alias dans la liste, faites-en le tout premier nom de backend ʻavailable, et si aucun d'entre eux n'est ʻavailable, faites-en une chaîne vide.

Lorsque la conversion de nom est terminée, regardez le backend correspondant au nom converti, renvoyez le nom s'il y en a un et lancez une exception dans le cas contraire.

Lisez la classe parent get_backend ()

Après avoir converti le nom et obtenu un nom «disponible»,

        return super().get_backend(name=name, **kwargs)

Je disais. Alors lisons BaseProvider.get_backend (). Omettez la docstring.

    def get_backend(self, name=None, **kwargs):
        backends = self.backends(name, **kwargs)
        if len(backends) > 1:
            raise QiskitBackendNotFoundError('More than one backend matches the criteria')
        if not backends:
            raise QiskitBackendNotFoundError('No backend matches the criteria')

        return backends[0]

Renvoie le premier élément de la liste ou quelque chose retourné par BasicAerProvider.backends (name).

Lisez BasicAerProvider.backends.

    def backends(self, name=None, filters=None, **kwargs):
        # pylint: disable=arguments-differ
        backends = self._backends.values()

        # Special handling of the `name` parameter, to support alias resolution
        # and deprecated names.
        if name:
            try:
                resolved_name = resolve_backend_name(
                    name, backends,
                    self._deprecated_backend_names(),
                    {}
                )
                backends = [backend for backend in backends if
                            backend.name() == resolved_name]
            except LookupError:
                return []

        return filter_backends(backends, filters=filters, **kwargs)

Il appelle également resolution_backend_name. À moins que «obsolète» ne circule ou quelque chose d'étrange comme ça, vous devriez obtenir le même résultat, peu importe le nombre de fois que vous essayez. Cette fois, nous extrayons le backend lui-même, pas le nom.

Quant à filter_backends, c'est long même si je n'ai pas fait grand-chose, donc si vous êtes intéressé, qiskit / fournisseurs / fournisseursutils.py Veuillez vous référer à /qiskit/providers/providerutils.py).

Seuls les docstring et les commentaires sont cités.

def filter_backends(backends, filters=None, **kwargs):
    """Abréviation
    Args:
        backends (list[BaseBackend]): list of backends.
        filters (callable): filtering conditions as a callable.
        **kwargs: dict of criteria.
    Returns:
        list[BaseBackend]: a list of backend instances matching the
            conditions.
    """
    # Inspect the backends to decide which filters belong to
    # backend.configuration and which ones to backend.status, as it does
    # not involve querying the API.

    # 1. Apply backend.configuration filtering.
    # 2. Apply backend.status filtering (it involves one API call for
    # each backend).
    # 3. Apply acceptor filter.

kwargs donne les conditions pour backend.configuration et backend.status. Si ceux-ci sont donnés de manière mixte, ils seront triés du côté filter_backends. De plus, si vous donnez la fonction à l'argument filters, il retournera le résultat du filtrage avec la fonction filter intégrée à Python.

Au fait, l'histoire tourne mal. Vous pouvez passer None comme argument à la fonction filter intégrée à Python. Je ne savais pas.

Return an iterator yielding those items of iterable for which function(item) is true. If function is None, return the items that are true.

Il semble.

list(filter(None, [1, 2, 0, 3, "", "a"]))                                                        
# => [1, 2, 3, 'a']

Quoi qu'il en soit, j'ai pu obtenir un backend qui correspondait au nom, bien qu'il soit foiré. avec ça

backend_sim = BasicAer.get_backend('qasm_simulator')

Vous pouvez lire la ligne.

Lire ʻexecute`

continuer

result = execute(qc, backend_sim).result()

Dans la ligne de, lisez ʻexecute`. Depuis qiskit / execute.py, docstring est supprimé et cité.

def execute(experiments, backend,
            basis_gates=None, coupling_map=None,  # circuit transpile options
            backend_properties=None, initial_layout=None,
            seed_transpiler=None, optimization_level=None, pass_manager=None,
            qobj_id=None, qobj_header=None, shots=1024,  # common run options
            memory=False, max_credits=10, seed_simulator=None,
            default_qubit_los=None, default_meas_los=None,  # schedule run options
            schedule_los=None, meas_level=2, meas_return='avg',
            memory_slots=None, memory_slot_size=100, rep_time=None, parameter_binds=None,
            **run_config):
    # transpiling the circuits using given transpile options
    experiments = transpile(experiments,
                            basis_gates=basis_gates,
                            coupling_map=coupling_map,
                            backend_properties=backend_properties,
                            initial_layout=initial_layout,
                            seed_transpiler=seed_transpiler,
                            optimization_level=optimization_level,
                            backend=backend,
                            pass_manager=pass_manager,
                            )

    # assembling the circuits into a qobj to be run on the backend
    qobj = assemble(experiments,
                    qobj_id=qobj_id,
                    qobj_header=qobj_header,
                    shots=shots,
                    memory=memory,
                    max_credits=max_credits,
                    seed_simulator=seed_simulator,
                    default_qubit_los=default_qubit_los,
                    default_meas_los=default_meas_los,
                    schedule_los=schedule_los,
                    meas_level=meas_level,
                    meas_return=meas_return,
                    memory_slots=memory_slots,
                    memory_slot_size=memory_slot_size,
                    rep_time=rep_time,
                    parameter_binds=parameter_binds,
                    backend=backend,
                    **run_config
                    )

    # executing the circuits on the backend and returning the job
    return backend.run(qobj, **run_config)

Regardez transpile

Cette fois, je n'entrerai pas dans les détails de Transpile. Je ne regarde vraiment que la surface.

Il peut être trouvé sur qiskit / compiler / transpile.py. Je vais l'omettre car la docstring est très longue, mais c'est intéressant et je recommande de la lire.

def transpile(circuits,
              backend=None,
              basis_gates=None, coupling_map=None, backend_properties=None,
              initial_layout=None, seed_transpiler=None,
              optimization_level=None,
              pass_manager=None, callback=None, output_name=None):
    # transpiling schedules is not supported yet.
    if isinstance(circuits, Schedule) or \
            (isinstance(circuits, list) and all(isinstance(c, Schedule) for c in circuits)):
        return circuits

    if optimization_level is None:
        config = user_config.get_config()
        optimization_level = config.get('transpile_optimization_level', None)

    # Get TranspileConfig(s) to configure the circuit transpilation job(s)
    circuits = circuits if isinstance(circuits, list) else [circuits]
    transpile_configs = _parse_transpile_args(circuits, backend, basis_gates, coupling_map,
                                              backend_properties, initial_layout,
                                              seed_transpiler, optimization_level,
                                              pass_manager, callback, output_name)
    # Check circuit width against number of qubits in coupling_map(s)
    coupling_maps_list = list(config.coupling_map for config in transpile_configs)
    for circuit, parsed_coupling_map in zip(circuits, coupling_maps_list):
        # If coupling_map is not None
        if isinstance(parsed_coupling_map, CouplingMap):
            n_qubits = len(circuit.qubits)
            max_qubits = parsed_coupling_map.size()
            if n_qubits > max_qubits:
                raise TranspilerError('Number of qubits ({}) '.format(n_qubits) +
                                      'in {} '.format(circuit.name) +
                                      'is greater than maximum ({}) '.format(max_qubits) +
                                      'in the coupling_map')
    # Transpile circuits in parallel
    circuits = parallel_map(_transpile_circuit, list(zip(circuits, transpile_configs)))

    if len(circuits) == 1:
        return circuits[0]
    return circuits

Dans les machines réelles, il y a des portes qui ne sont pas connectées à CNOT, il semble donc qu'elles travaillent à créer des circuits tout en les allouant. De plus, le calcul lui-même étant difficile, il est calculé en parallèle. Même si vous omettez l'argument, il récupérera les paramètres du back-end le cas échéant, grâce à l'ajout de nombreux paramètres au back-end.

Convertit le circuit et renvoie le circuit lui-même.

Regardez ʻassembler`

Regardez ʻassemble` dans qiskit / compiler / assemble.py de la même manière. (Docstring omis)

def assemble(experiments,
             backend=None,
             qobj_id=None, qobj_header=None,
             shots=1024, memory=False, max_credits=None, seed_simulator=None,
             qubit_lo_freq=None, meas_lo_freq=None,
             qubit_lo_range=None, meas_lo_range=None,
             schedule_los=None, meas_level=2, meas_return='avg', meas_map=None,
             memory_slot_size=100, rep_time=None, parameter_binds=None,
             **run_config):
    experiments = experiments if isinstance(experiments, list) else [experiments]
    qobj_id, qobj_header, run_config_common_dict = _parse_common_args(backend, qobj_id, qobj_header,
                                                                      shots, memory, max_credits,
                                                                      seed_simulator, **run_config)

    # assemble either circuits or schedules
    if all(isinstance(exp, QuantumCircuit) for exp in experiments):
        run_config = _parse_circuit_args(parameter_binds, **run_config_common_dict)

        # If circuits are parameterized, bind parameters and remove from run_config
        bound_experiments, run_config = _expand_parameters(circuits=experiments,
                                                           run_config=run_config)
        return assemble_circuits(circuits=bound_experiments, qobj_id=qobj_id,
                                 qobj_header=qobj_header, run_config=run_config)

    elif all(isinstance(exp, ScheduleComponent) for exp in experiments):
        run_config = _parse_pulse_args(backend, qubit_lo_freq, meas_lo_freq,
                                       qubit_lo_range, meas_lo_range,
                                       schedule_los, meas_level, meas_return,
                                       meas_map, memory_slot_size, rep_time,
                                       **run_config_common_dict)

        return assemble_schedules(schedules=experiments, qobj_id=qobj_id,
                                  qobj_header=qobj_header, run_config=run_config)

    else:
        raise QiskitError("bad input to assemble() function; "
                          "must be either circuits or schedules")

Il semble que vous puissiez assembler le circuit et le programme d'impulsions. Le type de valeur à renvoyer est «QObj», et dans le cas d'un circuit, le type «QasmQObj» est renvoyé.

Si vous ne savez pas quel type de structure de données est le type QasmQobj, vous aurez des problèmes après cela, donc

from qiskit import * 
qc = QuantumCircuit(2, 2) 
qc.h(0) 
qc.cx(0, 1) 
qc.measure([0,1], [0,1]) 
backend_sim = BasicAer.get_backend('qasm_simulator') 
tran = transpile(qc, backend=backend_sim)
asm = assemble(qc, backend=backend_sim)

En regardant «l'asm», c'était comme suit.

QasmQobj(
    config=QasmQobjConfig(
        memory=False,
        memory_slots=2,
        n_qubits=2,
        parameter_binds=[],
        shots=1024),
    experiments=[
        QasmQobjExperiment(
            config=QasmQobjExperimentConfig(memory_slots=2, n_qubits=2),
            header=QobjExperimentHeader(
                clbit_labels=[['c', 0], ['c', 1]],
                creg_sizes=[['c', 2]],
                memory_slots=2,
                n_qubits=2,
                name='circuit0',
                qreg_sizes=[['q', 2]],
                qubit_labels=[['q', 0], ['q', 1]]),
            instructions=[
                QasmQobjInstruction(
                    name='u2', params=[0, 3.14159265358979], qubits=[0]),
                QasmQobjInstruction(name='cx', qubits=[0, 1]),
                QasmQobjInstruction(memory=[0], name='measure', qubits=[0]),
                QasmQobjInstruction(memory=[1], name='measure', qubits=[1])
            ])
    ],
    header=QobjHeader(backend_name='qasm_simulator', backend_version='2.0.0'),
    qobj_id='06682d2e-bfd8-4dba-ba8e-4e46492c1609',
    schema_version='1.1.0',
    type='QASM')

Je pense que vous pouvez le lire sans aucun problème. A propos, la porte H est convertie en ʻu2 (0, π) `. C'est une expression équivalente.

Voir backend.run

Maintenant que nous avons converti le circuit en QasmQobj, ensuite

    return backend.run(qobj, **run_config)

Je lirai. qiskit / fournisseurs / basicaer / qasm_simulator.py

    def run(self, qobj, backend_options=None):
        """Run qobj asynchronously.
        Args:
            qobj (Qobj): payload of the experiment
            backend_options (dict): backend options
        Returns:
            BasicAerJob: derived from BaseJob
        Additional Information:
            backend_options: Is a dict of options for the backend. It may contain
                * "initial_statevector": vector_like
            The "initial_statevector" option specifies a custom initial
            initial statevector for the simulator to be used instead of the all
            zero state. This size of this vector must be correct for the number
            of qubits in all experiments in the qobj.
            Example::
                backend_options = {
                    "initial_statevector": np.array([1, 0, 0, 1j]) / np.sqrt(2),
                }
        """
        self._set_options(qobj_config=qobj.config,
                          backend_options=backend_options)
        job_id = str(uuid.uuid4())
        job = BasicAerJob(self, job_id, self._run_job, qobj)
        job.submit()
        return job

C'est bien de pouvoir mettre un premier vecteur dans les options. Il semble que cela dépend complètement du backend. L'option chop_threshold semble traiter la valeur comme 0 si elle est inférieure ou égale à cette valeur lors de la simulation du vecteur d'état.

Lire QasmSimulatorPy._set_options

Continuez à lire _set_options.

    def _set_options(self, qobj_config=None, backend_options=None):
        """Set the backend options for all experiments in a qobj"""
        # Reset default options
        self._initial_statevector = self.DEFAULT_OPTIONS["initial_statevector"]
        self._chop_threshold = self.DEFAULT_OPTIONS["chop_threshold"]
        if backend_options is None:
            backend_options = {}

        # Check for custom initial statevector in backend_options first,
        # then config second
        if 'initial_statevector' in backend_options:
            self._initial_statevector = np.array(backend_options['initial_statevector'],
                                                 dtype=complex)
        elif hasattr(qobj_config, 'initial_statevector'):
            self._initial_statevector = np.array(qobj_config.initial_statevector,
                                                 dtype=complex)
        if self._initial_statevector is not None:
            # Check the initial statevector is normalized
            norm = np.linalg.norm(self._initial_statevector)
            if round(norm, 12) != 1:
                raise BasicAerError('initial statevector is not normalized: ' +
                                    'norm {} != 1'.format(norm))
        # Check for custom chop threshold
        # Replace with custom options
        if 'chop_threshold' in backend_options:
            self._chop_threshold = backend_options['chop_threshold']
        elif hasattr(qobj_config, 'chop_threshold'):
            self._chop_threshold = qobj_config.chop_threshold

En raison du fait que le backend lui-même a des options

        # Reset default options
        self._initial_statevector = self.DEFAULT_OPTIONS["initial_statevector"]
        self._chop_threshold = self.DEFAULT_OPTIONS["chop_threshold"]

Est sorti dans un tel endroit. Quoi qu'il en soit, je viens d'interpréter les deux options et de les définir dans qobj.

Lisez BasicAerJob .__ init__

        job_id = str(uuid.uuid4())
        job = BasicAerJob(self, job_id, self._run_job, qobj)

Il a un UUID approprié comme job_id. Je fais de «BasicAerJob» un travail. self._run_job est une méthode, mais nous la transmettons telle quelle. Voyons-le lorsqu'il sera appelé plus tard.

qiskit/providers/basicaer/basicaerjob.py

class BasicAerJob(BaseJob):
    """BasicAerJob class.
    Attributes:
        _executor (futures.Executor): executor to handle asynchronous jobs
    """

    if sys.platform in ['darwin', 'win32']:
        _executor = futures.ThreadPoolExecutor()
    else:
        _executor = futures.ProcessPoolExecutor()

    def __init__(self, backend, job_id, fn, qobj):
        super().__init__(backend, job_id)
        self._fn = fn
        self._qobj = qobj
        self._future = None

Je viens de créer les données et je n'ai rien fait à ce stade. J'étais un peu inquiet à propos de l'utilisation du pool de threads sur Windows et Mac et du pool de processus sur d'autres. En multi-threading, même s'il y a plusieurs processeurs, un seul thread peut s'exécuter en même temps. Faisons attention à la façon dont il est utilisé.

Lisez BasicAerJob.submit

        job.submit()
        return job

Je soumets un emploi puis je retourne un emploi. Lisons BasicAerJob.submit.

    def submit(self):
        """Submit the job to the backend for execution.
        Raises:
            QobjValidationError: if the JSON serialization of the Qobj passed
            during construction does not validate against the Qobj schema.
            JobError: if trying to re-submit the job.
        """
        if self._future is not None:
            raise JobError("We have already submitted the job!")

        validate_qobj_against_schema(self._qobj)
        self._future = self._executor.submit(self._fn, self._job_id, self._qobj)
        validate_qobj_against_schema(self._qobj)

Quant à, Qobj correspond-il au format du schéma JSON défini dans un autre fichier? A été uniquement vérifié à l'aide de la bibliothèque externe jsonschema. Ce n'est pas très intéressant, je vais donc l'omettre.

_executor était le ThreadPoolExecutor ou ProcessPoolExecutor dans la bibliothèque standard Pythonconcurrent.futures. Nous appelons cette méthode «submit».

Lisez les Documents officiels.

submit(fn, *args, **kwargs) Planifiez l'exécution de l'objet appelable fn en tant que fn (* args ** kwargs) et représentez l'exécution de l'objet appelable [Future](https://docs.python.org/en /3/library/concurrent.futures.html#concurrent.futures.Future) Renvoie un objet.

En d'autres termes, vous créez un objet Future pour programmer self._fn (self._job_id, self._qobj) à exécuter.

Cet objet Future est pour une exécution asynchrone, et tout en étant déplacé dans les coulisses par _executor, le processus se poursuivra même si le calcul n'est pas terminé. (Si vous commandez un hamburger dans un magasin de hamburgers, vous pouvez attendre devant le comptoir jusqu'à ce qu'il soit terminé, mais si vous obtenez une étiquette numérotée, vous pouvez d'abord vous asseoir ou faire autre chose et passer votre temps de manière significative. Avec les appels de fonction normaux, vous devez attendre que le résultat revienne, mais avec une exécution asynchrone, vous pouvez faire autre chose sans attendre qu'il revienne. L'objet Future recevra le hamburger dans le futur. C'est comme une étiquette numérique

Lire QasmSimulatorPy._run_job

Rappelant ce qui était self._fn (self._job_id, self._qobj) piloté par _executor,self made withjob = BasicAerJob (self, job_id, self._run_job, qobj)C'était ._run_job (job_id, qobj).

Alors lisons QasmSimulatorPy._run_job.

    def _run_job(self, job_id, qobj):
        """Run experiments in qobj
        Args:
            job_id (str): unique id for the job.
            qobj (Qobj): job description
        Returns:
            Result: Result object
        """
        self._validate(qobj)
        result_list = []
        self._shots = qobj.config.shots
        self._memory = getattr(qobj.config, 'memory', False)
        self._qobj_config = qobj.config
        start = time.time()
        for experiment in qobj.experiments:
            result_list.append(self.run_experiment(experiment))
        end = time.time()
        result = {'backend_name': self.name(),
                  'backend_version': self._configuration.backend_version,
                  'qobj_id': qobj.qobj_id,
                  'job_id': job_id,
                  'results': result_list,
                  'status': 'COMPLETED',
                  'success': True,
                  'time_taken': (end - start),
                  'header': qobj.header.to_dict()}

        return Result.from_dict(result)

--self._validate (lire plus tard) --Définissez les informations nécessaires sur le backend lui-même --Démarrer la minuterie

Lire QasmSimulatorPy._validate

    def _validate(self, qobj):
        """Semantic validations of the qobj which cannot be done via schemas."""
        n_qubits = qobj.config.n_qubits
        max_qubits = self.configuration().n_qubits
        if n_qubits > max_qubits:
            raise BasicAerError('Number of qubits {} '.format(n_qubits) +
                                'is greater than maximum ({}) '.format(max_qubits) +
                                'for "{}".'.format(self.name()))
        for experiment in qobj.experiments:
            name = experiment.header.name
            if experiment.config.memory_slots == 0:
                logger.warning('No classical registers in circuit "%s", '
                               'counts will be empty.', name)
            elif 'measure' not in [op.name for op in experiment.instructions]:
                logger.warning('No measurements in circuit "%s", '
                               'classical register will remain all zeros.', name)

Nous vérifions que le nombre de qubits ne dépasse pas la valeur maximale et qu'il existe un registre classique et que la mesure est en cours. (Pas de registre classique, aucune mesure n'est traitée comme un avertissement, pas une erreur)

Afficher QasmSimulatorPy.run_experiment

Avant cela, rappelons-nous de quoi les «expériences» étaient faites.

    experiments=[
        QasmQobjExperiment(
            config=QasmQobjExperimentConfig(memory_slots=2, n_qubits=2),
            header=QobjExperimentHeader(
                clbit_labels=[['c', 0], ['c', 1]],
                creg_sizes=[['c', 2]],
                memory_slots=2,
                n_qubits=2,
                name='circuit0',
                qreg_sizes=[['q', 2]],
                qubit_labels=[['q', 0], ['q', 1]]),
            instructions=[
                QasmQobjInstruction(
                    name='u2', params=[0, 3.14159265358979], qubits=[0]),
                QasmQobjInstruction(name='cx', qubits=[0, 1]),
                QasmQobjInstruction(memory=[0], name='measure', qubits=[0]),
                QasmQobjInstruction(memory=[1], name='measure', qubits=[1])
            ])
    ],

«QasmQobjExperiment» correspond à un circuit quantique. Avec Qiskit, vous pouvez emballer et exécuter plusieurs circuits quantiques, mais cette fois, il semble qu'il n'y en ait qu'un. Puisque nous exécutions la boucle de for experiment dans qobj.experiments:, nous appelons run_experiment pour chaque circuit. Le circuit a des "instructions", qui correspondent à des portes et des mesures.

run_experiment est vraiment long, je vais donc l'omettre. En gros, la porte est calculée et la mesure est calculée pour le nombre de «coups».

        return {'name': experiment.header.name,
                'seed_simulator': seed_simulator,
                'shots': self._shots,
                'data': data,
                'status': 'DONE',
                'success': True,
                'time_taken': (end - start),
                'header': experiment.header.to_dict()}

Renvoie un résultat comme. De plus, «data» est

        # Add data
        data = {'counts': dict(Counter(memory))}
        # Optionally add memory list
        if self._memory:
            data['memory'] = memory
        # Optionally add final statevector
        if self.SHOW_FINAL_STATE:
            data['statevector'] = self._get_statevector()
            # Remove empty counts and memory for statevector simulator
            if not data['counts']:
                data.pop('counts')
            if 'memory' in data and not data['memory']

Comme, il a des données de mesure.

Lisez qiskit.result.Result

Le type de dictionnaire est changé en type «Résultat» par «return Result.from_dict (result)». qiskit/result/result.py

@bind_schema(ResultSchema)
class Result(BaseModel):
    """Model for Results.
    Please note that this class only describes the required fields. For the
    full description of the model, please check ``ResultSchema``.
    Attributes:
        backend_name (str): backend name.
        backend_version (str): backend version, in the form X.Y.Z.
        qobj_id (str): user-generated Qobj id.
        job_id (str): unique execution id from the backend.
        success (bool): True if complete input qobj executed correctly. (Implies
            each experiment success)
        results (list[ExperimentResult]): corresponding results for array of
            experiments of the input qobj
    """

    def __init__(self, backend_name, backend_version, qobj_id, job_id, success,
                 results, **kwargs):
        self.backend_name = backend_name
        self.backend_version = backend_version
        self.qobj_id = qobj_id
        self.job_id = job_id
        self.success = success
        self.results = results

À l'origine, je veux lire from_dict, mais l'une des fonctionnalités de qiskit-terra est que je fais de mon mieux pour traiter le schéma JSON. from_dict est défini dans la classe parente BaseModel, et son contenu est

    @classmethod
    def from_dict(cls, dict_):
        """Deserialize a dict of simple types into an instance of this class.
        Note that this method requires that the model is bound with
        ``@bind_schema``.
        """
        try:
            data = cls.schema.load(dict_)
        except ValidationError as ex:
            raise ModelValidationError(
                ex.messages, ex.field_name, ex.data, ex.valid_data, **ex.kwargs) from None

        return data

C'est très simple. Créez une classe ResultSchema séparément de la classe Result

class ResultSchema(BaseSchema):
    """Schema for Result."""

    # Required fields.
    backend_name = String(required=True)
    backend_version = String(required=True,
                             validate=Regexp('[0-9]+.[0-9]+.[0-9]+$'))
    qobj_id = String(required=True)
    job_id = String(required=True)
    success = Boolean(required=True)
    results = Nested(ExperimentResultSchema, required=True, many=True)

    # Optional fields.
    date = DateTime()
    status = String()
    header = Nested(ObjSchema)

Et ainsi de suite, je vais plutôt bien. (Défini à /qiskit/result/models.py) Je ne peux pas le lire correctement, mais je pense que ʻExperimentResultSchema` etc. sera récursivement transformé en objets tout en confirmant qu'ils ont ces types de données.

Dans tous les cas, il s'agit d'une recharge de données, alors laissez-moi l'omettre cette fois.

Le «résultat» que j'ai déplacé et rendu à portée de main était le suivant.

Result(
    backend_name='qasm_simulator',
    backend_version='2.0.0',
    header=Obj(backend_name='qasm_simulator', backend_version='2.0.0'),
    job_id='65b162ae-fe6b-480a-85ba-8c890d8bbf3b',
    qobj_id='c75e9b2c-da7c-4dd6-96da-d132126e6dc0',
    results=[
        ExperimentResult(
            data=ExperimentResultData(counts=Obj(0x0=493, 0x3=531)),
            header=Obj(
                clbit_labels=[['c', 0], ['c', 1]],
                creg_sizes=[['c', 2]],
                memory_slots=2,
                n_qubits=2,
                name='circuit0',
                qreg_sizes=[['q', 2]],
                qubit_labels=[['q', 0], ['q', 1]]),
            meas_level=2,
            name='circuit0',
            seed_simulator=1480888590,
            shots=1024,
            status='DONE',
            success=True,
            time_taken=0.12355875968933105)
    ],
    status='COMPLETED',
    success=True,
    time_taken=0.12369537353515625)

Lisez BasicAerJob.result ()

J'ai sauté le milieu, mais

result = execute(qc, backend_sim).result()

Quand ʻexecute (qc, backend_sim) est terminé, nous regarderons .result () . ʻExecute a renvoyé BasicAerJob.

    @requires_submit
    def result(self, timeout=None):
        # pylint: disable=arguments-differ
        """Get job result. The behavior is the same as the underlying
        concurrent Future objects,
        https://docs.python.org/3/library/concurrent.futures.html#future-objects
        Args:
            timeout (float): number of seconds to wait for results.
        Returns:
            qiskit.Result: Result object
        Raises:
            concurrent.futures.TimeoutError: if timeout occurred.
            concurrent.futures.CancelledError: if job cancelled before completed.
        """
        return self._future.result(timeout=timeout)

Le premier @ requires_submit voit si vous avez fait unesubmit. Qu'il soit ou non soumis a été défini sur self._future = None lorsque le BasicAerJob a été créé, mais lorsque soumis, l'objet Future est inclus dans self._future, donc il est réglé sur None Vous pouvez le savoir en vérifiant si ce n'est pas le cas.

Cela appelle normalement la méthode "result" de l'objet "Future". Voir la documentation officielle (https://docs.python.org/ja/3/library/concurrent.futures.html#concurrent.futures.Future).

result(timeout=None) Renvoie la valeur renvoyée par l'appel. Si l'appel n'est pas encore terminé, cette méthode attend quelques secondes d'expiration. Si l'appel ne se termine pas dans les secondes du délai d'attente, une erreur concurrent.futures.TimeoutError est levée. Vous pouvez spécifier int ou float pour le délai d'expiration. Si timeout n'est pas spécifié ou est None, il n'y a pas de limite au temps d'attente. CanceledError est levé s'il est annulé avant la fin du futur. Si l'appel lève une exception, cette méthode lève la même exception.

C'est une méthode qui renvoie le résultat si elle est terminée, attend si elle n'est pas terminée, puis renvoie le résultat. Le résultat est un objet de type «Result» comme vu ci-dessus.

Lisez Result.get_counts

Enfin la dernière ligne.

print(result.get_counts(qc))

Je regarderais.

    def get_counts(self, experiment=None):
        """Get the histogram data of an experiment.
        Args:
            experiment (str or QuantumCircuit or Schedule or int or None): the index of the
                experiment, as specified by ``get_data()``.
        Returns:
            dict[str:int]: a dictionary with the counts for each qubit, with
                the keys containing a string in binary format and separated
                according to the registers in circuit (e.g. ``0100 1110``).
                The string is little-endian (cr[0] on the right hand side).
        Raises:
            QiskitError: if there are no counts for the experiment.
        """
        exp = self._get_experiment(experiment)
        try:
            header = exp.header.to_dict()
        except (AttributeError, QiskitError):  # header is not available
            header = None

        if 'counts' in self.data(experiment).keys():
            return postprocess.format_counts(self.data(experiment)['counts'],
                                             header)
        elif 'statevector' in self.data(experiment).keys():
            vec = postprocess.format_statevector(self.data(experiment)['statevector'])
            return state_to_counts(vec)
        else:
            raise QiskitError('No counts for experiment "{0}"'.format(experiment))

Dans _get_experiment, ʻexperiment` correspondant au nom du circuit est supprimée. S'il n'y a qu'une seule «expérience» comme cette fois, vous pouvez omettre l'argument «expérience».

Si vous pouvez extraire l'en-tête de ʻexperiment, vous connaîtrez la longueur du registre classique, vous pourrez donc bien remplir le résultat de la mesure à zéro. Dans format_counts`, je fais une belle conversion en une chaîne de caractères et je la place dans un dictionnaire.

Résumé

Cette fois, avec Qiskit, j'ai lu le flux de traitement sauf pour la partie directement liée au calcul quantique. En raison de changements répétés et d'une évolutivité à l'épreuve du temps, certaines pièces avaient des spécifications très difficiles, mais elles ont été conçues pour pouvoir faire beaucoup de choses, et elles ont également été converties en transpile et JSON, etc. J'ai eu l'impression que les parties que Blueqat ne traite pas sont assez lourdes et ce sera une belle expérience d'apprentissage.

C'est la fin de mon objectif de lire tout le processus, mais je continuerai à lire le code source de Qiskit à l'avenir.

Recommended Posts

Lecture du code source Qiskit ~ Terra: lecture du backend Get, Call, Get Result
Lecture du code source Qiskit ~ Terra: lecture de la création du circuit à l'ajout de portes et de mesures
[Python] Lire le code source de Flask