Skip to content

qek.data.extractors

docs module qek.data.extractors

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
"""
High-Level API to compile raw data (graphs) and process it on a quantum device, either a local emulator,
a remote emulator or a physical QPI.
"""

import abc
import asyncio
from dataclasses import dataclass
import itertools
import json
import logging
from math import ceil
from uuid import UUID
import time
from typing import Any, Callable, Generator, Generic, Sequence, TypeVar, cast
from numpy.typing import NDArray
from pasqal_cloud import SDK
from pasqal_cloud.device import BaseConfig, EmuTNConfig, EmulatorType
from pasqal_cloud.job import Job
from pasqal_cloud.utils.filters import JobFilters
from pathlib import Path
import numpy as np
import os
import pulser as pl
from pulser.devices import Device
from pulser.json.abstract_repr.deserializer import deserialize_device
from pulser_simulation import QutipEmulator
from torch.utils.data import Dataset

from qek.data import processed_data
from qek.data.graphs import BaseGraph, BaseGraphCompiler
from qek.data.processed_data import ProcessedData
from qek.shared.error import CompilationError

logger = logging.getLogger(__name__)


@dataclass
class Compiled:
    """
    The result of compiling a graph for execution on a quantum device.
    """

    # Future plans: as of this writing, this class (or a reworked version of it)
    # is expected to move to the `qool-layer` library.

    # The graph itself.
    graph: BaseGraph

    # A sequence adapted to the quantum device.
    sequence: pl.Sequence


@dataclass
class Feature:
    """
    A feature extracted from raw data.
    """

    data: NDArray[np.floating]


class BaseExtracted(abc.ABC):
    """
    Data extracted by one of the subclasses of `BaseExtractor`.

    Note that the list of processed data will generally *not* contain all the graphs ingested
    by the Extractor, as not all graphs may not be compiled for a given device.
    """

    def __init__(self, device: Device):
        self.device = device

    def __await__(self) -> Generator[Any, Any, None]:
        """
        Wait asynchronously until execution is ready.

        This will avoid blocking your main thread, so calling this method once,
        before the first call to `processed_data`, is strongly recommended
        for use on a server or an interactive application.
        """
        # By default, no need to wait.
        yield None

    @property
    @abc.abstractmethod
    def processed_data(self) -> list[processed_data.ProcessedData]:
        pass

    @property
    @abc.abstractmethod
    def raw_data(self) -> list[BaseGraph]:
        """
        A subset of the graphs ingested by the Extractor.
        """
        pass

    @property
    @abc.abstractmethod
    def targets(self) -> list[int] | None:
        """
        If available, the machine-learning targets for these graphs, in the same order and with the same number of entrie as `raw_data`.
        """
        pass

    @property
    def states(self) -> list[dict[str, int]]:
        """
        The quantum states extracted from `raw_data` by executing `sequences` on the device, in the same order and with the same number of entries as `raw_data`.
        """
        return [data.state_dict for data in self.processed_data]

    def features(self, size_max: int | None) -> list[Feature]:
        """
        The features extracted from `raw_data` by processing `states`, in the same order and with the same number of entries as `raw_data`.

        By default, the features extracted are the distribution of excitation levels based on `states`. However, subclasses may override
        this method to provide custom features extraction.

        Arguments:
            size_max (optional) Performance/precision lever. If specified, specifies the number of qubits to take into account from all
                the `states`. If `size_max` is lower than the number of qubits used to extract `self.states[i]` (i.e. the number of qubits
                in `self.sequences[i]`), then only take into account the `size_max` first qubits of this state to extract
                `self.features(size_max)[i]`. If, on the other hand, `size_max` is greater than the number of qubits used to extract
                `self.states[i]`, pad `self.features(size_max)[i]` with 0s.
                If unspecified, use the largest number of qubits in `selfsequences`.
        """
        if size_max is None:
            for data in self.processed_data:
                seq = data._sequence
                if size_max is None or len(seq.qubit_info) > size_max:
                    size_max = len(seq.qubit_info)
        if size_max is None:
            # The only way size_max can be None is if `self.sequences` is empty.
            return []

        return [Feature(processed_data.dist_excitation(state, size_max)) for state in self.states]

    def save_dataset(self, file_path: Path) -> None:
        """Saves the processed dataset to a JSON file.

        Note: This does NOT attempt to save the graphs.

        Args:
            dataset: The dataset to be saved.
            file_path: The path where the dataset will be saved as a JSON
                file.

        Note:
            The data is stored in a format suitable for loading with load_dataset.
        """
        with open(file_path, "w") as file:
            states = self.states
            targets = self.targets
            data = [
                {
                    "sequence": self.processed_data[i]._sequence.to_abstract_repr(),
                    # Some emulators will actually be `dict[str, int64]` instead of `dict[str, int]` and `int64`
                    # is not JSON-serializable.
                    #
                    # The reason for which `int64` is not JSON-serializable is that JSON limits ints to 2^53-1.
                    # However, in practice, this should not be a problem, since the `int`/`int64` in our dict is
                    # limited to the number of runs, and we don't expect to be launching 2^53 consecutive runs
                    # for a single sequence on a device in any foreseeable future (assuming a run of 1ns,
                    # this would still take ~4 billion years to execute).
                    "state_dict": {key: int(value) for (key, value) in states[i].items()},
                    "target": targets[i] if targets is not None else None,
                }
                for i in range(len(self.processed_data))
            ]
            json.dump(data, file)
        logger.info("processed data saved to %s", file_path)


class SyncExtracted(BaseExtracted):
    """
    Data extracted synchronously, i.e. no need to wait for a remote server.
    """

    def __init__(
        self,
        raw_data: list[BaseGraph],
        targets: list[int] | None,
        sequences: list[pl.Sequence],
        states: list[dict[str, int]],
    ):
        assert len(raw_data) == len(sequences)
        assert len(sequences) == len(states)
        if targets is not None:
            if len(targets) < len(sequences):
                # Not all graphs come with a target.
                #
                # This Extracted will not be usable as the training sample, so ignore all targets.
                if len(targets) != 0:
                    logger.debug(
                        "We compiled %s graphs but we only have %s targets, ignoring all targets",
                        len(sequences),
                        len(targets),
                    )
                targets = None
        self._raw_data = raw_data
        self._targets = targets
        self._sequences = sequences
        self._states = states
        self._processed_data = [
            ProcessedData(
                sequence=seq, state_dict=cast(dict[str, int | np.int64], state), target=target
            )
            for (seq, state, target) in itertools.zip_longest(sequences, states, targets or [])
        ]

    @property
    def processed_data(self) -> list[ProcessedData]:
        return self._processed_data

    @property
    def raw_data(self) -> list[BaseGraph]:
        return self._raw_data

    @property
    def targets(self) -> list[int] | None:
        return self._targets

    @property
    def sequences(self) -> list[pl.Sequence]:
        return self._sequences

    @property
    def states(self) -> list[dict[str, int]]:
        return self._states


# Type variable for BaseExtractor[GraphType].
GraphType = TypeVar("GraphType")


class BaseExtractor(abc.ABC, Generic[GraphType]):
    """
    The base of the hierarchy of extractors.

    The role of extractors is to take a list of raw data (here, labelled graphs) into
    processed data containing machine-learning features (here, excitation vectors).

    Args:
        path: If specified, the processed data will be saved to this file as JSON once
            the execution is complete.
        device: A quantum device for which the data should be prepared.
        compiler: A graph compiler, in charge of converting graphs to Pulser Sequences,
            the format that can be executed on a quantum device.
    """

    def __init__(
        self, device: Device, compiler: BaseGraphCompiler[GraphType], path: Path | None = None
    ) -> None:
        self.path = path

        # The list of graphs (raw data). Fill it with `self.add_graphs`.
        self.graphs: list[BaseGraph] = []
        self.device: Device = device

        # The compiled sequences. Filled with `self.compile`.
        # Note that the list of compiled sequences may be shorter than the list of
        # raw data, as not all graphs may be compiled to a given `device`.
        self.sequences: list[Compiled] = []
        self.compiler = compiler

        # A counter used to give a unique id to each graph.
        self._counter = 0

    def save(self, snapshot: list[ProcessedData]) -> None:
        """Saves a dataset to a JSON file.

        Args:
            dataset (list[ProcessedData]): The dataset to be saved, containing
                RegisterData instances.
            file_path (str): The path where the dataset will be saved as a JSON
                file.

        Note:
            The data is stored in a format suitable for loading with load_dataset.
        """
        if self.path is not None:
            with open(self.path, "w") as file:
                data = [
                    {
                        "sequence": instance._sequence.to_abstract_repr(),
                        "state_dict": instance.state_dict,
                        "target": instance.target,
                    }
                    for instance in snapshot
                ]
                json.dump(data, file)
            logger.info("processed data saved to %s", self.path)

    def compile(
        self, filter: Callable[[BaseGraph, pl.Sequence, int], bool] | None = None
    ) -> list[Compiled]:
        """
        Compile all pending graphs into Pulser sequences that the Quantum Device may execute.

        Once this method has succeeded, the results are stored in `self.sequences`.
        """
        if len(self.graphs) == 0:
            raise Exception("No graphs to compile, did you forget to call `import_graphs`?")
        if filter is None:
            filter = lambda _graph, sequence, _index: True  # noqa: E731
        self.sequences = []
        for graph in self.graphs:
            try:
                register = graph.compile_register()
                pulse = graph.compile_pulse()
                sequence = pl.Sequence(register=register.register, device=graph.device)
                sequence.declare_channel("ising", "rydberg_global")
                sequence.add(pulse.pulse, "ising")
            except CompilationError as e:
                # In some cases, we produce graphs that pass `is_embeddable` but cannot be compiled.
                # It _looks_ like this is due to rounding errors. We're investigating this in issue #29,
                # but for the time being, we're simply logging and skipping them.
                logger.debug("Graph #%s could not be compiled (%s), skipping", graph.id, e)
                continue
            if not filter(graph, sequence, graph.id):
                logger.debug("Graph #%s did not pass filtering, skipping", graph.id)
                continue
            logger.debug("Compiling graph #%s for execution on the device", graph.id)
            self.sequences.append(Compiled(graph=graph, sequence=sequence))
        logger.debug("Compilation step complete, %s graphs compiled", len(self.sequences))
        return self.sequences

    def add_graphs(self, graphs: Sequence[GraphType] | Dataset[GraphType]) -> None:
        """
        Add new graphs to compile and run.
        """
        for graph in graphs:
            self._counter += 1
            id = self._counter
            logger.debug("ingesting # %s", id)
            processed = self.compiler.ingest(graph=graph, device=self.device, id=id)
            # Skip graphs that are not embeddable.
            if processed.is_embeddable():
                logger.debug("graph # %s is embeddable, accepting", id)
                self.graphs.append(processed)
            else:
                logger.info("graph # %s is not embeddable, skipping", id)
        logger.info("imported %s graphs", len(self.graphs))

    @abc.abstractmethod
    def run(self) -> BaseExtracted:
        """
        Run compiled graphs.

        You will need to call `self.compile` first, to make sure that the graphs are compiled.

        Returns:
            Data extracted by this extractor.

            Not all extractors may return the same data, so please take a look at the documentation
            of the extractor you are using.
        """
        raise Exception("Not implemented")


class QutipExtractor(BaseExtractor[GraphType]):
    """
    A Extractor that uses the Qutip Emulator to run sequences compiled
    from graphs.

    Performance note: emulating a quantum device on a classical
    computer requires considerable amount of resources, so this
    Extractor may be slow or require too much memory.

    See also:
    - EmuMPSExtractor (alternative emulator, generally much faster)
    - QPUExtractor (run on a physical QPU)

    Args:
        path: Path to store the result of the run, for future uses.
            To reload the result of a previous run, use `LoadExtractor`.
        compiler: A graph compiler, in charge of converting graphs to Pulser Sequences,
            the format that can be executed on a quantum device.
        device: A device to use. For general experiments, the default
            device `AnalogDevice` is a perfectly reasonable choice.
    """

    def __init__(
        self,
        compiler: BaseGraphCompiler[GraphType],
        device: Device = pl.devices.AnalogDevice,
        path: Path | None = None,
    ):
        super().__init__(path=path, device=device, compiler=compiler)
        self.graphs: list[BaseGraph]
        self.device = device

    def run(self, max_qubits: int = 8) -> SyncExtracted:
        """
        Run the compiled graphs.

        As emulating a quantum device is slow consumes resources and time exponential in the
        number of qubits, for the sake of performance, we limit the number of qubits in the execution
        of this extractor.

        Args:
            max_qubits: Skip any sequence that require strictly more than `max_qubits`. Defaults to 8.

        Returns:
            Processed data for all the sequences that were executed.
        """
        if len(self.sequences) == 0:
            logger.warning("No sequences to run, did you forget to call compile()?")
            return SyncExtracted(raw_data=[], targets=[], sequences=[], states=[])

        raw_data: list[BaseGraph] = []
        targets: list[int] = []
        sequences: list[pl.Sequence] = []
        states: list[dict[str, int]] = []
        for compiled in self.sequences:
            qubits_used = len(compiled.sequence.qubit_info)
            if qubits_used > max_qubits:
                logger.info(
                    "Graph %s exceeds the qubit limit specified in QutipExtractor (%s > %s), skipping",
                    id,
                    qubits_used,
                    max_qubits,
                )
                continue
            logger.debug("Executing compiled graph # %s", id)
            simul = QutipEmulator.from_sequence(sequence=compiled.sequence)
            counter = cast(dict[str, Any], simul.run().sample_final_state())
            logger.debug("Execution of compiled graph # %s complete", id)
            raw_data.append(compiled.graph)
            if compiled.graph.target is not None:
                targets.append(compiled.graph.target)
            sequences.append(compiled.sequence)
            states.append(counter)

        result = SyncExtracted(
            raw_data=raw_data, targets=targets, sequences=sequences, states=states
        )
        logger.debug("Emulation step complete, %s compiled graphs executed", len(raw_data))
        if self.path is not None:
            result.save_dataset(self.path)
        return result


if os.name == "posix":
    # Any Unix including Linux and macOS

    import emu_mps

    class EmuMPSExtractor(BaseExtractor[GraphType]):
        """
        A Extractor that uses the emu-mps Emulator to run sequences compiled
        from graphs.

        Performance note: emulating a quantum device on a classical
        computer requires considerable amount of resources, so this
        Extractor may be slow or require too much memory. If should,
        however, be faster than QutipExtractor in most cases.

        See also:
        - QPUExtractor (run on a physical QPU)

        Args:
            path: Path to store the result of the run, for future uses.
                To reload the result of a previous run, use `LoadExtractor`.
            compiler: A graph compiler, in charge of converting graphs to Pulser Sequences,
                the format that can be executed on a quantum device.
            device: A device to use. For general experiments, the default
                device `AnalogDevice` is a perfectly reasonable choice.
        """

        def __init__(
            self,
            compiler: BaseGraphCompiler[GraphType],
            device: Device = pl.devices.AnalogDevice,
            path: Path | None = None,
        ):
            super().__init__(device=device, compiler=compiler, path=path)
            self.graphs: list[BaseGraph]
            self.device = device

        def run(self, max_qubits: int = 10, dt: int = 10) -> BaseExtracted:
            """
            Run the compiled graphs.

            As emulating a quantum device is slow consumes resources and time exponential in the
            number of qubits, for the sake of performance, we limit the number of qubits in the execution
            of this extractor.

            Args:
                max_qubits: Skip any sequence that require strictly more than `max_qubits`. Defaults to 8.
                dt: The duration of the simulation step, in us. Defaults to 10.

            Returns:
                Processed data for all the sequences that were executed.
            """
            if len(self.sequences) == 0:
                logger.warning("No sequences to run, did you forget to call compile()?")
                return SyncExtracted(raw_data=[], targets=[], sequences=[], states=[])

            backend = emu_mps.MPSBackend()
            raw_data = []
            targets: list[int] = []
            sequences = []
            states = []
            for compiled in self.sequences:
                qubits_used = len(compiled.sequence.qubit_info)
                if qubits_used > max_qubits:
                    logger.info(
                        "Graph %s exceeds the qubit limit specified in EmuMPSExtractor (%s > %s), skipping",
                        id,
                        qubits_used,
                        max_qubits,
                    )
                    continue
                logger.debug("Executing compiled graph # %s", id)

                # Configure observable.
                cutoff_duration = int(ceil(compiled.sequence.get_duration() / dt) * dt)
                observable = emu_mps.BitStrings(evaluation_times={cutoff_duration})
                config = emu_mps.MPSConfig(observables=[observable], dt=dt)
                counter: dict[str, Any] = backend.run(compiled.sequence, config)[observable.name][
                    cutoff_duration
                ]
                logger.debug("Execution of compiled graph # %s complete", id)
                raw_data.append(compiled.graph)
                if compiled.graph.target is not None:
                    targets.append(compiled.graph.target)
                sequences.append(compiled.sequence)
                states.append(counter)

            logger.debug("Emulation step complete, %s compiled graphs executed", len(raw_data))

            result = SyncExtracted(
                raw_data=raw_data, targets=targets, sequences=sequences, states=states
            )
            logger.debug("Emulation step complete, %s compiled graphs executed", len(raw_data))
            if self.path is not None:
                result.save_dataset(self.path)
            return result


# How many seconds to sleep while waiting for the results from the cloud.
SLEEP_DELAY_S = 2


class PasqalCloudExtracted(BaseExtracted):
    """
    Data extracted from the cloud API, i.e. we need wait for a remote server.

    Performance note:
        If your code is meant to be executed as part of an interactive application or
        a server, you should consider calling `await extracted` before your first call
        to any of the methods of `extracted`. Otherwise, you will block the main thread.

        If you are running this as part of an experiment, a Jupyter notebook, etc. you
        do not need to do so.
    """

    def __init__(
        self,
        compiled: list[Compiled],
        job_ids: list[str],
        sdk: SDK,
        state_extractor: Callable[[Job, pl.Sequence], dict[str, int] | None],
        path: Path | None = None,
    ):
        """
        Prepare for reception of data.

        Arguments:
            compiled: The result of compiling a set of graphs.
            job_ids: The ids of the jobs on the cloud API, in the same order as `compiled`.
            state_extractor: A callback used to extract the counter from a job.
                Used as various cloud back-ends return different formats.
            path: If provided, a path at which to save the results once they're available.
        """
        self._compiled = compiled
        self._job_ids = job_ids
        self._results: SyncExtracted | None = None
        self._path = path
        self._sdk = sdk
        self._state_extractor = state_extractor

    def _wait(self) -> None:
        """
        Wait synchronously until remote execution is ready.

        This WILL BLOCK your main thread, possibly for a very long time.
        """
        if self._results is not None:
            # Results are already available.
            return
        pending_job_ids: set[str] = set(self._job_ids)
        completed_jobs: dict[str, Job] = {}
        while len(pending_job_ids) > 0:
            time.sleep(SLEEP_DELAY_S)

            # Fetch up to 100 pending jobs (upstream limits).
            MAX_JOB_LEN = 100
            check_ids: list[str | UUID] = [cast(str | UUID, id) for id in pending_job_ids][
                :MAX_JOB_LEN
            ]

            # Update their status.
            check_jobs = self._sdk.get_jobs(filters=JobFilters(id=check_ids))
            for job in check_jobs.results:
                assert isinstance(job, Job)
                if job.status not in {"PENDING", "RUNNING"}:
                    logger.debug("Job %s is now complete", job.id)
                    pending_job_ids.discard(job.id)
                    completed_jobs[job.id] = job

        # At this point, all jobs are complete.
        self._ingest(completed_jobs)

    def __await__(self) -> Generator[Any, Any, None]:
        """
        Wait asynchronously until remote execution is ready.

        This will NOT block your main thread, so this method is strongly recommended
        for use on a server or an interactive application.

        Example:
            await extracted
        """
        if self._results is not None:
            # Results are already available.
            return
        pending_job_ids: set[str] = set(self._job_ids)
        completed_jobs: dict[str, Job] = {}
        while len(pending_job_ids) > 0:
            yield from asyncio.sleep(SLEEP_DELAY_S).__await__()

            # Fetch up to 100 pending jobs (upstream limits).
            MAX_JOB_LEN = 100
            check_ids: list[str | UUID] = [cast(str | UUID, id) for id in pending_job_ids][
                :MAX_JOB_LEN
            ]

            # Update their status.
            check_jobs = self._sdk.get_jobs(
                filters=JobFilters(id=check_ids)
            )  # Ideally, this should be async, see https://github.com/pasqal-io/pasqal-cloud/issues/162.
            for job in check_jobs.results:
                assert isinstance(job, Job)
                if job.status not in {"PENDING", "RUNNING"}:
                    logger.debug("Job %s is now complete", job.id)
                    pending_job_ids.discard(job.id)
                    completed_jobs[job.id] = job

        # At this point, all jobs are complete.
        self._ingest(completed_jobs)

    def _ingest(self, jobs: dict[str, Job]) -> None:
        """
        Ingest data received from the remote server.

        No I/O.
        """
        assert len(jobs) == len(self._job_ids)

        raw_data = []
        targets: list[int] = []
        sequences = []
        states = []
        for i, id in enumerate(self._job_ids):
            job = jobs[id]
            compiled = self._compiled[i]
            if job.status == "DONE":
                state_dict = self._state_extractor(job, compiled.sequence)
                if state_dict is None:
                    logger.warning(
                        "Job %s (graph %s) did not return a usable state, skipping",
                        i,
                        compiled.graph.id,
                    )
                    continue
                raw_data.append(compiled.graph)
                if compiled.graph.target is not None:
                    targets.append(compiled.graph.target)
                sequences.append(compiled.sequence)
                states.append(state_dict)
            else:
                # If some sequences failed, let's skip them and proceed as well as we can.
                logger.warning(
                    "Job %s (graph %s) failed with status %s and errors %s, skipping",
                    i,
                    compiled.graph.id,
                    job.status,
                    job.errors,
                )
        self._results = SyncExtracted(
            raw_data=raw_data, targets=targets, sequences=sequences, states=states
        )
        if self._path is not None:
            self.save_dataset(self._path)

    @property
    def processed_data(self) -> list[ProcessedData]:
        self._wait()
        assert self._results is not None
        return self._results.processed_data

    @property
    def raw_data(self) -> list[BaseGraph]:
        self._wait()
        assert self._results is not None
        return self._results.raw_data

    @property
    def targets(self) -> list[int] | None:
        self._wait()
        assert self._results is not None
        return self._results.targets

    @property
    def sequences(self) -> list[pl.Sequence]:
        self._wait()
        assert self._results is not None
        return self._results.sequences

    @property
    def states(self) -> list[dict[str, int]]:
        self._wait()
        assert self._results is not None
        return self._results.states


class BaseRemoteExtractor(BaseExtractor[GraphType], Generic[GraphType]):
    """
    An Extractor that uses a remote Quantum Device published
    on Pasqal Cloud, to run sequences compiled from graphs.

    Performance note (servers and interactive applications only):
        If your code is meant to be executed as part of an interactive application or
        a server, you should consider calling `await extracted` before your first call
        to any of the methods of `extracted`. Otherwise, you will block the main thread.

        If you are running this as part of an experiment, a Jupyter notebook, etc. you
        may ignore this performance note.

    Args:
        path: Path to store the result of the run, for future uses.
            To reload the result of a previous run, use `LoadExtractor`.
        project_id: The ID of the project on the Pasqal Cloud API.
        username: Your username on the Pasqal Cloud API.
        password: Your password on the Pasqal Cloud API. If you leave
            this to None, you will need to enter your password manually.
        device_name: The name of the device to use. As of this writing,
            the default value of "FRESNEL" represents the latest QPU
            available through the Pasqal Cloud API.
        job_id: Use this to resume a workflow e.g. after turning off
            your computer while the QPU was executing your sequences.
            Warning: A job started with one executor MUST NOT be resumed
            with a different executor.
    """

    def __init__(
        self,
        compiler: BaseGraphCompiler[GraphType],
        project_id: str,
        username: str,
        device_name: str,
        password: str | None = None,
        job_ids: list[str] | None = None,
        path: Path | None = None,
    ):
        sdk = SDK(username=username, project_id=project_id, password=password)

        # Fetch the latest list of QPUs
        specs = sdk.get_device_specs_dict()
        device = cast(Device, deserialize_device(specs[device_name]))

        super().__init__(device=device, compiler=compiler, path=path)
        self._sdk = sdk
        self._job_ids: list[str] | None = job_ids

    @property
    def job_ids(self) -> list[str] | None:
        return self._job_ids

    @abc.abstractmethod
    def run(
        self,
    ) -> PasqalCloudExtracted:
        """
        Launch the extraction.
        """
        raise NotImplementedError()

    def _run(
        self,
        state_extractor: Callable[[Job, pl.Sequence], dict[str, int] | None],
        emulator: EmulatorType | None,
        config: BaseConfig | None,
    ) -> PasqalCloudExtracted:
        if len(self.sequences) == 0:
            logger.warning("No sequences to run, did you forget to call compile()?")
            return PasqalCloudExtracted(
                compiled=[],
                job_ids=[],
                sdk=self._sdk,
                path=self.path,
                state_extractor=state_extractor,
            )

        device: pl.devices.Device = self.sequences[0].sequence.device
        # As of this writing, the API doesn't support runs longer than 500 jobs.
        # If we want to add more runs, we'll need to split them across several jobs.
        max_runs = device.max_runs if isinstance(device.max_runs, int) else 500

        if self._job_ids is None:
            # Enqueue jobs.
            self._job_ids = []
            for compiled in self.sequences:
                logger.debug("Enqueuing execution of compiled graph #%s", compiled.graph.id)
                job = self._sdk.create_batch(
                    compiled.sequence.to_abstract_repr(),
                    jobs=[{"runs": max_runs}],
                    wait=False,
                    emulator=emulator,
                    configuration=config,
                )
                assert len(job.ordered_jobs) == 1
                job_id = job.ordered_jobs[0].id
                logger.info(
                    "Remote execution of compiled graph #%s starting, job with id %s",
                    compiled.graph.id,
                    job_id,
                )
                self._job_ids.append(job_id)
            logger.info(
                "All %s jobs enqueued for remote execution, with ids %s",
                len(self._job_ids),
                self._job_ids,
            )
        assert len(self._job_ids) == len(self.sequences)

        return PasqalCloudExtracted(
            compiled=self.sequences,
            job_ids=self._job_ids,
            sdk=self._sdk,
            path=self.path,
            state_extractor=state_extractor,
        )


class RemoteQPUExtractor(BaseRemoteExtractor[GraphType]):
    """
    An Extractor that uses a remote QPU published
    on Pasqal Cloud, to run sequences compiled from graphs.

    Performance note:
        as of this writing, the waiting lines for a QPU
        may be very long. You may use this Extractor to resume your workflow
        with a computation that has been previously started.

    Performance note (servers and interactive applications only):
        If your code is meant to be executed as part of an interactive application or
        a server, you should consider calling `await extracted` before your first call
        to any of the methods of `extracted`. Otherwise, you will block the main thread.

        If you are running this as part of an experiment, a Jupyter notebook, etc. you
        may ignore this performance note.

    Args:
        path: Path to store the result of the run, for future uses.
            To reload the result of a previous run, use `LoadExtractor`.
        project_id: The ID of the project on the Pasqal Cloud API.
        username: Your username on the Pasqal Cloud API.
        password: Your password on the Pasqal Cloud API. If you leave
            this to None, you will need to enter your password manually.
        device_name: The name of the device to use. As of this writing,
            the default value of "FRESNEL" represents the latest QPU
            available through the Pasqal Cloud API.
        job_id: Use this to resume a workflow e.g. after turning off
            your computer while the QPU was executing your sequences.
    """

    def __init__(
        self,
        compiler: BaseGraphCompiler[GraphType],
        project_id: str,
        username: str,
        device_name: str = "FRESNEL",
        password: str | None = None,
        job_ids: list[str] | None = None,
        path: Path | None = None,
    ):
        super().__init__(
            compiler=compiler,
            project_id=project_id,
            username=username,
            device_name=device_name,
            password=password,
            job_ids=job_ids,
            path=path,
        )

    def run(self) -> PasqalCloudExtracted:
        return self._run(emulator=None, config=None, state_extractor=lambda job, _seq: job.result)


class RemoteEmuMPSExtractor(BaseRemoteExtractor[GraphType]):
    """
    An Extractor that uses a remote high-performance emulator (EmuMPS)
    published on Pasqal Cloud, to run sequences compiled from graphs.

    Performance note (servers and interactive applications only):
        If your code is meant to be executed as part of an interactive application or
        a server, you should consider calling `await extracted` before your first call
        to any of the methods of `extracted`. Otherwise, you will block the main thread.

        If you are running this as part of an experiment, a Jupyter notebook, etc. you
        may ignore this performance note.

    Args:
        path: Path to store the result of the run, for future uses.
            To reload the result of a previous run, use `LoadExtractor`.
        project_id: The ID of the project on the Pasqal Cloud API.
        username: Your username on the Pasqal Cloud API.
        password: Your password on the Pasqal Cloud API. If you leave
            this to None, you will need to enter your password manually.
        device_name: The name of the device to use. As of this writing,
            the default value of "FRESNEL" represents the latest QPU
            available through the Pasqal Cloud API.
        job_id: Use this to resume a workflow e.g. after turning off
            your computer while the QPU was executing your sequences.
    """

    def __init__(
        self,
        compiler: BaseGraphCompiler[GraphType],
        project_id: str,
        username: str,
        device_name: str = "FRESNEL",
        password: str | None = None,
        job_ids: list[str] | None = None,
        path: Path | None = None,
    ):
        super().__init__(
            compiler=compiler,
            project_id=project_id,
            username=username,
            device_name=device_name,
            password=password,
            job_ids=job_ids,
            path=path,
        )

    def run(self, dt: int = 10) -> PasqalCloudExtracted:
        def extractor(job: Job, sequence: pl.Sequence) -> dict[str, int] | None:
            full_result = job.full_result
            if full_result is None:
                return None
            result = full_result["counter"]
            if result is None:
                return None
            assert isinstance(result, dict)
            return result

        return self._run(
            emulator=EmulatorType.EMU_MPS,
            config=EmuTNConfig(
                dt=dt,
            ),
            state_extractor=extractor,
        )