Skip to content

Quera

CustomSubmissionRoutine

Bases: RoutineBase

submit

submit(
    shots,
    url,
    json_body_template,
    method="POST",
    args=(),
    request_options={},
    use_experimental=False,
    sleep_time=0.1,
)

Compile to QuEraTaskSpecification and submit to a custom service.

Parameters:

Name Type Description Default
shots int

number of shots

required
url str

url of the custom service

required
json_body_template str

json body template, must contain '{task_ir}'

required
method str

http method to be used. Defaults to "POST".

'POST'
args Tuple[LiteralType]

additional arguments to be passed into the

()
request_options Dict[str, Any]

additional options to be passed into the request method,

{}
use_experimental bool

Enable experimental hardware capabilities

False
sleep_time float

time to sleep between each request. Defaults to 0.1.

0.1

Returns:

Type Description
List[Tuple[NamedTuple, Response]]

List[Tuple[NamedTuple, Response]]: List of parameters for each batch in

List[Tuple[NamedTuple, Response]]

the task and the response from the post request.

Examples:

Here is a simple example of how to use this method. Note the body_template has double curly braces on the outside to escape the string formatting.

>>> body_template = "{{"token": "my_token", "task": {task_ir}}}"
>>> responses = (
    program.quera.custom.submit(
        100,
        "http://my_custom_service.com",
        body_template
    )
)
Source code in src/bloqade/ir/routine/quera.py
def submit(
    self,
    shots: int,
    url: str,
    json_body_template: str,
    method: str = "POST",
    args: Tuple[LiteralType] = (),
    request_options: Dict[str, Any] = {},
    use_experimental: bool = False,
    sleep_time: float = 0.1,
) -> List[Tuple[NamedTuple, Response]]:
    """Compile to QuEraTaskSpecification and submit to a custom service.

    Args:
        shots (int): number of shots
        url (str): url of the custom service
        json_body_template (str): json body template, must contain '{task_ir}'
        which is a placeholder for a string representation of the task ir.
        The task ir string will be inserted into the template with
        `json_body_template.format(task_ir=task_ir_string)`.
        to be replaced by QuEraTaskSpecification
        method (str): http method to be used. Defaults to "POST".
        args (Tuple[LiteralType]): additional arguments to be passed into the
        compiler coming from `args` option of the build. Defaults to ().
        request_options: additional options to be passed into the request method,
        Note the `data` option will be overwritten by the
        `json_body_template.format(task_ir=task_ir_string)`.
        use_experimental (bool): Enable experimental hardware capabilities
        sleep_time (float): time to sleep between each request. Defaults to 0.1.

    Returns:
        List[Tuple[NamedTuple, Response]]: List of parameters for each batch in
        the task and the response from the post request.

    Examples:
        Here is a simple example of how to use this method. Note the body_template
        has double curly braces on the outside to escape the string formatting.

    ```python
    >>> body_template = "{{"token": "my_token", "task": {task_ir}}}"
    >>> responses = (
        program.quera.custom.submit(
            100,
            "http://my_custom_service.com",
            body_template
        )
    )
    ```
    """

    if r"{task_ir}" not in json_body_template:
        raise ValueError(r"body_template must contain '{task_ir}'")

    partial_eval = json_body_template.format(task_ir='"task_ir"')
    try:
        _ = json.loads(partial_eval)
    except json.JSONDecodeError as e:
        raise ValueError(
            "body_template must be a valid json template. "
            'When evaluating template with task_ir="task_ir", '
            f"the template evaluated to: {partial_eval!r}.\n"
            f"JSONDecodeError: {e}"
        )

    out = []
    for metadata, task_ir in self._compile(shots, use_experimental, args):
        json_request_body = json_body_template.format(
            task_ir=task_ir.json(exclude_none=True, exclude_unset=True)
        )
        request_options.update(data=json_request_body)
        response = request(method, url, **request_options)
        out.append((metadata, response))
        time.sleep(sleep_time)

    return out

QuEraHardwareRoutine

Bases: RoutineBase

run_async

run_async(
    shots,
    args=(),
    name=None,
    use_experimental=False,
    shuffle=False,
    **kwargs
)

Compile to a RemoteBatch, which contain QuEra backend specific tasks, and run_async through QuEra service.

Parameters:

Name Type Description Default
shots int

number of shots

required
args Tuple

additional arguments

()
name str

custom name of the batch

None
shuffle bool

shuffle the order of jobs

False
Return

RemoteBatch

Source code in src/bloqade/ir/routine/quera.py
@beartype
def run_async(
    self,
    shots: int,
    args: Tuple[LiteralType, ...] = (),
    name: Optional[str] = None,
    use_experimental: bool = False,
    shuffle: bool = False,
    **kwargs,
) -> RemoteBatch:
    """
    Compile to a RemoteBatch, which contain
        QuEra backend specific tasks,
        and run_async through QuEra service.

    Args:
        shots (int): number of shots
        args (Tuple): additional arguments
        name (str): custom name of the batch
        shuffle (bool): shuffle the order of jobs

    Return:
        RemoteBatch

    """
    batch = self._compile(shots, use_experimental, args, name)
    batch._submit(shuffle, **kwargs)
    return batch