Skip to content

Bloqade

BloqadeEmulation dataclass 

BloqadeEmulation(
    task_data: TaskData,
    compile_cache: Optional[CompileCache] = None,
)

Data class to hold the Hamiltonian and metadata for a given set of parameters

hamiltonian property 

hamiltonian: RydbergHamiltonian

Return the Hamiltonian object for the given task data.

metadata property 

metadata: NamedTuple

The metadata for the given task data.

evolve 

evolve(
    state: Optional[StateVector] = None,
    solver_name: str = "dop853",
    atol: float = 1e-07,
    rtol: float = 1e-14,
    nsteps: int = 2147483647,
    times: Sequence[float] = (),
    interaction_picture: bool = False,
) -> Iterator[StateVector]

Evolve an initial state vector using the Hamiltonian

Parameters:

Name Type Description Default
state Optional[StateVector]

The initial state vector to

 None
solver_name str

Which SciPy Solver to use. Defaults to

 'dop853'
atol float

Absolute tolerance for ODE solver. Defaults

 1e-07
rtol float

Relative tolerance for adaptive step in

 1e-14
nsteps int

Maximum number of steps allowed per integration

 2147483647
times Sequence[float]

The times to evaluate the state vector

 ()
interaction_picture bool

Use the interaction picture when

 False

Returns:

Type Description
Iterator[StateVector]

Iterator[StateVector]: An iterator of the state vectors at each time step.
Source code in .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/bloqade.py 
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
def evolve(
    self,
    state: Optional[StateVector] = None,
    solver_name: str = "dop853",
    atol: float = 1e-7,
    rtol: float = 1e-14,
    nsteps: int = 2147483647,
    times: Sequence[float] = (),
    interaction_picture: bool = False,
) -> Iterator[StateVector]:
    """Evolve an initial state vector using the Hamiltonian

    Args:
        state (Optional[StateVector], optional): The initial state vector to
        evolve. if not provided, the zero state will be used. Defaults to None.
        solver_name (str, optional): Which SciPy Solver to use. Defaults to
        "dop853".
        atol (float, optional): Absolute tolerance for ODE solver. Defaults
        to 1e-14.
        rtol (float, optional): Relative tolerance for adaptive step in
        ODE solver. Defaults to 1e-7.
        nsteps (int, optional): Maximum number of steps allowed per integration
        step. Defaults to 2147483647.
        times (Sequence[float], optional): The times to evaluate the state vector
        at. Defaults to (). If not provided the state will be evaluated at
        the end of the bloqade program.
        interaction_picture (bool, optional): Use the interaction picture when
        solving schrodinger equation. Defaults to False.

    Returns:
        Iterator[StateVector]: An iterator of the state vectors at each time step.

    """
    state = self.zero_state(np.complex128) if state is None else state

    U = AnalogGate(self.hamiltonian)

    return U.apply(
        state,
        times=times,
        solver_name=solver_name,
        atol=atol,
        rtol=rtol,
        nsteps=nsteps,
        interaction_picture=interaction_picture,
    )

fock_state 

fock_state(
    fock_state_str: str, dtype: dtype = np.float64
) -> StateVector

Return the fock state for the given Hamiltonian.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/bloqade.py 
91
92
93
94
95
96
97
98
def fock_state(
    self, fock_state_str: str, dtype: np.dtype = np.float64
) -> StateVector:
    """Return the fock state for the given Hamiltonian."""
    index = self.hamiltonian.space.fock_state_to_index(fock_state_str)
    data = np.zeros(self.hamiltonian.space.size, dtype=dtype)
    data[index] = 1
    return StateVector(data, self.hamiltonian.space)

zero_state 

zero_state(dtype: dtype = np.float64) -> StateVector

Return the zero state for the given Hamiltonian.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/bloqade.py 
87
88
89
def zero_state(self, dtype: np.dtype = np.float64) -> StateVector:
    """Return the zero state for the given Hamiltonian."""
    return self.hamiltonian.space.zero_state(dtype)

BloqadePythonRoutine

Bases: RoutineBase


              flowchart TD
              bloqade.analog.ir.routine.bloqade.BloqadePythonRoutine[BloqadePythonRoutine]
              bloqade.analog.ir.routine.base.RoutineBase[RoutineBase]
              bloqade.analog.ir.routine.base.RoutineParse[RoutineParse]
              bloqade.analog.builder.parse.trait.Parse[Parse]
              bloqade.analog.builder.parse.trait.ParseRegister[ParseRegister]
              bloqade.analog.builder.parse.trait.ParseSequence[ParseSequence]
              bloqade.analog.builder.parse.trait.ParseCircuit[ParseCircuit]
              bloqade.analog.builder.parse.trait.ParseRoutine[ParseRoutine]
              bloqade.analog.ir.routine.base.RoutineShow[RoutineShow]
              bloqade.analog.builder.parse.trait.Show[Show]

                              bloqade.analog.ir.routine.base.RoutineBase --> bloqade.analog.ir.routine.bloqade.BloqadePythonRoutine
                                bloqade.analog.ir.routine.base.RoutineParse --> bloqade.analog.ir.routine.base.RoutineBase
                                bloqade.analog.builder.parse.trait.Parse --> bloqade.analog.ir.routine.base.RoutineParse
                                bloqade.analog.builder.parse.trait.ParseRegister --> bloqade.analog.builder.parse.trait.Parse
                
                bloqade.analog.builder.parse.trait.ParseSequence --> bloqade.analog.builder.parse.trait.Parse
                
                bloqade.analog.builder.parse.trait.ParseCircuit --> bloqade.analog.builder.parse.trait.Parse
                
                bloqade.analog.builder.parse.trait.ParseRoutine --> bloqade.analog.builder.parse.trait.Parse
                


                bloqade.analog.ir.routine.base.RoutineShow --> bloqade.analog.ir.routine.base.RoutineBase
                                bloqade.analog.builder.parse.trait.Show --> bloqade.analog.ir.routine.base.RoutineShow
                




              click bloqade.analog.ir.routine.bloqade.BloqadePythonRoutine href "" "bloqade.analog.ir.routine.bloqade.BloqadePythonRoutine"
              click bloqade.analog.ir.routine.base.RoutineBase href "" "bloqade.analog.ir.routine.base.RoutineBase"
              click bloqade.analog.ir.routine.base.RoutineParse href "" "bloqade.analog.ir.routine.base.RoutineParse"
              click bloqade.analog.builder.parse.trait.Parse href "" "bloqade.analog.builder.parse.trait.Parse"
              click bloqade.analog.builder.parse.trait.ParseRegister href "" "bloqade.analog.builder.parse.trait.ParseRegister"
              click bloqade.analog.builder.parse.trait.ParseSequence href "" "bloqade.analog.builder.parse.trait.ParseSequence"
              click bloqade.analog.builder.parse.trait.ParseCircuit href "" "bloqade.analog.builder.parse.trait.ParseCircuit"
              click bloqade.analog.builder.parse.trait.ParseRoutine href "" "bloqade.analog.builder.parse.trait.ParseRoutine"
              click bloqade.analog.ir.routine.base.RoutineShow href "" "bloqade.analog.ir.routine.base.RoutineShow"
              click bloqade.analog.builder.parse.trait.Show href "" "bloqade.analog.builder.parse.trait.Show"

hamiltonian 

hamiltonian(
    *args: LiteralType,
    blockade_radius: float = 0.0,
    use_hyperfine: bool = False,
    waveform_runtime: str = "interpret",
    cache_matrices: bool = False
) -> List[BloqadeEmulation]

Generates a list of BloqadeEmulation objects which contain the Hamiltonian of your program.

If you have a variable(s) in your program you have assigned multiple values via batch_assign() there will be multiple BloqadeEmulation objects, one for each value. On the other hand if the program only assumes a singular value per each variable, there will only be one BloqadeEmulation object but it will still be encapsulated in a list.

Parameters:

Name Type Description Default
*args LiteralType

If your program has a variable that was declared as run-time assignable via .args you may pass a value to it here. If there are multiple variables declared via .args the order in which you assign values to those variables through this argument should follow the order in which the declaration occurred.

 ()
blockade_radius float

The radius in which atoms blockade eachother. Default value is 0.0 micrometers.

 0.0
use_hyperfine bool

Should the Hamiltonian account for hyperfine levels. Default value is False.

 False
waveform_runtime str

Specify which runtime to use for waveforms. If "numba" is specify the waveform is compiled, otherwise it is interpreted via the "interpret" argument. Defaults to "interpret".

 'interpret'
cache_matrices bool

Speed up Hamiltonian generation by reusing data (when possible) from previously generated Hamiltonians. Default value is False.

 False

Returns:

Type Description
List[BloqadeEmulation]

List[BloqadeEmulation]
Source code in .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/bloqade.py 
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
def hamiltonian(
    self,
    *args: LiteralType,
    blockade_radius: float = 0.0,
    use_hyperfine: bool = False,
    waveform_runtime: str = "interpret",
    cache_matrices: bool = False,
) -> List[BloqadeEmulation]:
    """
    Generates a list of BloqadeEmulation objects which contain the Hamiltonian of your program.

    If you have a variable(s) in your program you have assigned multiple values via `batch_assign()`
    there will be multiple `BloqadeEmulation` objects, one for each value. On the other hand
    if the program only assumes a singular value per each variable, there will only be
    one `BloqadeEmulation` object but it will still be encapsulated in a list.


    Args:
        *args (LiteralType): If your program has a variable that was declared as run-time assignable
            via `.args` you may pass a value to it here. If there are multiple
            variables declared via `.args` the order in which you assign values to those variables
            through this argument should follow the order in which the declaration occurred.
        blockade_radius (float): The radius in which atoms blockade eachother. Default value is 0.0 micrometers.
        use_hyperfine (bool): Should the Hamiltonian account for hyperfine levels. Default value is False.
        waveform_runtime (str): Specify which runtime to use for waveforms. If "numba" is specify the waveform
            is compiled, otherwise it is interpreted via the "interpret" argument. Defaults to "interpret".
        cache_matrices (bool): Speed up Hamiltonian generation by reusing data (when possible) from previously generated Hamiltonians.
            Default value is False.

    Returns:
        List[BloqadeEmulation]

    """
    ir_iter = self._generate_ir(
        args, blockade_radius, waveform_runtime, use_hyperfine
    )

    if cache_matrices:
        compile_cache = CompileCache()
    else:
        compile_cache = None

    return [
        BloqadeEmulation(task_data, compile_cache=compile_cache)
        for task_data in ir_iter
    ]

run 

run(
    shots: int,
    args: Tuple[LiteralType, ...] = (),
    name: Optional[str] = None,
    blockade_radius: float = 0.0,
    waveform_runtime: str = "interpret",
    interaction_picture: bool = False,
    cache_matrices: bool = False,
    multiprocessing: bool = False,
    num_workers: Optional[int] = None,
    solver_name: str = "dop853",
    atol: float = 1e-07,
    rtol: float = 1e-14,
    nsteps: int = 2147483647,
) -> LocalBatch

Run the current program using bloqade python backend

Parameters:

Name Type Description Default
shots int

number of shots after running state vector simulation

 required
args Tuple[LiteralType, ...]

The values for parameters defined

 ()
name Optional[str]

Name to give this run. Defaults to None.

 None
blockade_radius float

Use the Blockade subspace given a

 0.0
waveform_runtime str

(bool, optional): Use Numba to compile the waveforms,

 'interpret'
interaction_picture bool

Use the interaction picture when

 False
cache_matrices bool

Reuse previously evaluated matrcies when

 False
multiprocessing bool

Use multiple processes to process the

 False
num_workers Optional[int]

Number of processes to run with

 None
solver_name str

Which SciPy Solver to use. Defaults to

 'dop853'
atol float

Absolute tolerance for ODE solver. Defaults to

 1e-07
rtol float

Relative tolerance for adaptive step in ODE solver.

 1e-14
nsteps int

Maximum number of steps allowed per integration

 2147483647

Raises:

Type Description
ValueError

Cannot use multiprocessing and cache_matrices at the same time.

Returns:

Name Type Description
LocalBatch LocalBatch

Batch of local tasks that have been executed.
Source code in .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/bloqade.py 
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
@beartype
def run(
    self,
    shots: int,
    args: Tuple[LiteralType, ...] = (),
    name: Optional[str] = None,
    blockade_radius: float = 0.0,
    waveform_runtime: str = "interpret",
    interaction_picture: bool = False,
    cache_matrices: bool = False,
    multiprocessing: bool = False,
    num_workers: Optional[int] = None,
    solver_name: str = "dop853",
    atol: float = 1e-7,
    rtol: float = 1e-14,
    nsteps: int = 2_147_483_647,
) -> LocalBatch:
    """Run the current program using bloqade python backend

    Args:
        shots (int): number of shots after running state vector simulation
        args (Tuple[LiteralType, ...], optional): The values for parameters defined
        in `args`. Defaults to ().
        name (Optional[str], optional): Name to give this run. Defaults to None.
        blockade_radius (float, optional): Use the Blockade subspace given a
        particular radius. Defaults to 0.0.
        waveform_runtime: (bool, optional): Use Numba to compile the waveforms,
        Defaults to False.
        interaction_picture (bool, optional): Use the interaction picture when
        solving schrodinger equation. Defaults to False.
        cache_matrices (bool, optional): Reuse previously evaluated matrcies when
        possible. Defaults to False.
        multiprocessing (bool, optional): Use multiple processes to process the
        batches. Defaults to False.
        num_workers (Optional[int], optional): Number of processes to run with
        multiprocessing. Defaults to None.
        solver_name (str, optional): Which SciPy Solver to use. Defaults to
        "dop853".
        atol (float, optional): Absolute tolerance for ODE solver. Defaults to
        1e-14.
        rtol (float, optional): Relative tolerance for adaptive step in ODE solver.
        Defaults to 1e-7.
        nsteps (int, optional): Maximum number of steps allowed per integration
        step. Defaults to 2_147_483_647, the maximum value.

    Raises:
        ValueError: Cannot use multiprocessing and cache_matrices at the same time.

    Returns:
        LocalBatch: Batch of local tasks that have been executed.
    """
    if multiprocessing and cache_matrices:
        raise ValueError(
            "Cannot use multiprocessing and cache_matrices at the same time."
        )

    compile_options = dict(
        shots=shots,
        args=args,
        name=name,
        blockade_radius=blockade_radius,
        cache_matrices=cache_matrices,
        waveform_runtime=waveform_runtime,
    )

    solver_options = dict(
        multiprocessing=multiprocessing,
        num_workers=num_workers,
        solver_name=solver_name,
        atol=atol,
        rtol=rtol,
        nsteps=nsteps,
        interaction_picture=interaction_picture,
    )

    batch = self._compile(**compile_options)
    batch._run(**solver_options)

    return batch

run_callback 

run_callback(
    callback: Callable[
        [StateVector, NamedTuple, RydbergHamiltonian, Any],
        Any,
    ],
    program_args: Tuple[LiteralType, ...] = (),
    callback_args: Tuple = (),
    ignore_exceptions: bool = False,
    blockade_radius: float = 0.0,
    waveform_runtime: str = "interpret",
    interaction_picture: bool = False,
    cache_matrices: bool = False,
    multiprocessing: bool = False,
    num_workers: Optional[int] = None,
    solver_name: str = "dop853",
    atol: float = 1e-07,
    rtol: float = 1e-14,
    nsteps: int = 2147483647,
    use_hyperfine: bool = False,
) -> List

Run state-vector simulation with a callback to access full state-vector from emulator

Parameters:

Name Type Description Default
callback Callable[[StateVector, Metadata, RydbergHamiltonian, Any], Any]
required
program_args Tuple[LiteralType, ...]

The values for parameters

 ()
callback_args Tuple[Any, ...]

Extra arguments to pass into

 ()
ignore_exceptions bool

(bool, optional) If True any exception raised during

 False
blockade_radius float

Use the Blockade subspace given a

 0.0
waveform_runtime str

(str, optional): Specify which runtime to use for

 'interpret'
interaction_picture bool

Use the interaction picture when

 False
cache_matrices bool

Reuse previously evaluated matrcies when

 False
multiprocessing bool

Use multiple processes to process the

 False
num_workers Optional[int]

Number of processes to run with

 None
solver_name str

Which SciPy Solver to use. Defaults to

 'dop853'
atol float

Absolute tolerance for ODE solver. Defaults to

 1e-07
rtol float

Relative tolerance for adaptive step in ODE solver.

 1e-14
nsteps int

Maximum number of steps allowed per integration

 2147483647

Returns:

Name Type Description
List List

List of resulting outputs from the callbacks

Raises:

Type Description
RuntimeError

Raises the first error that occurs, only if
Note

For the callback function, first argument is the many-body wavefunction as a 1D complex numpy array, the second argument is of type Metadata which is a Named Tuple where the fields correspond to the parameters of that given task, RydbergHamiltonian is the object that contains the Hamiltonian used to generate the evolution for that task, Finally any optional positional arguments are allowed after that. The return value can be anything, the results will be collected in a list for each task in the batch.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/bloqade.py 
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
@beartype
def run_callback(
    self,
    callback: Callable[[StateVector, NamedTuple, RydbergHamiltonian, Any], Any],
    program_args: Tuple[LiteralType, ...] = (),
    callback_args: Tuple = (),
    ignore_exceptions: bool = False,
    blockade_radius: float = 0.0,
    waveform_runtime: str = "interpret",
    interaction_picture: bool = False,
    cache_matrices: bool = False,
    multiprocessing: bool = False,
    num_workers: Optional[int] = None,
    solver_name: str = "dop853",
    atol: float = 1e-7,
    rtol: float = 1e-14,
    nsteps: int = 2_147_483_647,
    use_hyperfine: bool = False,
) -> List:
    """Run state-vector simulation with a callback to access full state-vector from
    emulator

    Args:
        callback (Callable[[StateVector, Metadata, RydbergHamiltonian, Any], Any]):
        The callback function to run for each task in batch. See note below for more
        details about the signature of the function.
        program_args (Tuple[LiteralType, ...], optional): The values for parameters
        defined in `args`. Defaults to ().
        callback_args (Tuple[Any,...], optional): Extra arguments to pass into
        ignore_exceptions: (bool, optional) If `True` any exception raised during
        a task will be saved instead of the resulting output of the callback,
        otherwise the first exception by task number will be raised after *all*
        tasks have executed. Defaults to False.
        blockade_radius (float, optional): Use the Blockade subspace given a
        particular radius. Defaults to 0.0.
        waveform_runtime: (str, optional): Specify which runtime to use for
        waveforms. Defaults to "interpret".
        interaction_picture (bool, optional): Use the interaction picture when
        solving schrodinger equation. Defaults to False.
        cache_matrices (bool, optional): Reuse previously evaluated matrcies when
        possible. Defaults to False.
        multiprocessing (bool, optional): Use multiple processes to process the
        batches. Defaults to False.
        num_workers (Optional[int], optional): Number of processes to run with
        multiprocessing. Defaults to None.
        solver_name (str, optional): Which SciPy Solver to use. Defaults to
        "dop853".
        atol (float, optional): Absolute tolerance for ODE solver. Defaults to
        1e-14.
        rtol (float, optional): Relative tolerance for adaptive step in ODE solver.
        Defaults to 1e-7.
        nsteps (int, optional): Maximum number of steps allowed per integration
        step. Defaults to 2_147_483_647, the maximum value.

    Returns:
        List: List of resulting outputs from the callbacks

    Raises:
        RuntimeError: Raises the first error that occurs, only if
        `ignore_exceptions=False`.

    Note:
        For the `callback` function, first argument is the many-body wavefunction
        as a 1D complex numpy array, the second argument is of type `Metadata` which
        is a Named Tuple where the fields correspond to the parameters of that given
        task, RydbergHamiltonian is the object that contains the Hamiltonian used to
        generate the evolution for that task, Finally any optional positional
        arguments are allowed after that. The return value can be anything, the
        results will be collected in a list for each task in the batch.


    """
    if multiprocessing:
        from multiprocessing import Queue, Process, cpu_count
    else:
        from queue import Queue

    if cache_matrices:
        compile_cache = CompileCache()
    else:
        compile_cache = None

    solver_args = dict(
        solver_name=solver_name,
        atol=atol,
        rtol=rtol,
        nsteps=nsteps,
        interaction_picture=interaction_picture,
    )

    runner = self.EmuRunner(
        compile_cache=compile_cache,
        solver_args=solver_args,
        callback=callback,
        callback_args=callback_args,
    )

    tasks = Queue()
    results = Queue()

    total_tasks = 0
    ir_iter = self._generate_ir(
        program_args, blockade_radius, waveform_runtime, use_hyperfine
    )
    for task_data in ir_iter:
        task_number = task_data.task_id
        emulator_ir = task_data.emulator_ir
        metadata = task_data.metadata_dict
        total_tasks += 1
        tasks.put((task_number, (emulator_ir, metadata)))

    workers = []
    if multiprocessing:
        num_workers = max(int(num_workers or cpu_count()), 1)
        num_workers = min(total_tasks, num_workers)

        for _ in range(num_workers):
            worker = Process(
                target=BloqadePythonRoutine.process_tasks,
                args=(runner, tasks, results),
            )
            worker.start()

            workers.append(worker)
    else:
        self.process_tasks(runner, tasks, results)

    # blocks until all
    # results have been fetched
    # from the id_results Queue
    id_results = []
    for i in range(total_tasks):
        id_results.append(results.get())

    if workers:
        for worker in workers:
            worker.join()

        tasks.close()
        results.close()

    id_results.sort(key=lambda x: x[0])
    results = []

    for task_id, result in id_results:
        if not ignore_exceptions and isinstance(result, BaseException):
            try:
                raise result
            except BaseException:
                raise RuntimeError(
                    f"{result.__class__.__name__} occured during child process "
                    f"running for task number {task_id}:\n{traceback.format_exc()}"
                )

        results.append(result)

    return results

TaskData dataclass 

TaskData(
    task_id: int,
    emulator_ir: EmulatorProgram,
    metadata_dict: Dict[str, LiteralType],
)

Data class to hold the program ir and metadata for a given set of parameters