Quera

CustomSubmissionRoutine

Bases: RoutineBase

run

run(
    shots: int,
    RemoteTask: type[RemoteTaskType],
    args: Tuple[LiteralType, ...] = (),
    name: Optional[str] = None,
    use_experimental: bool = False,
    shuffle: bool = False,
    **kwargs
) -> RemoteBatch

Run the custom task and return the result.

Parameters:

Name	Type	Description	Default
shots	int	number of shots	required
RemoteTask	type	type of the remote task, must subclass of CustomRemoteTaskABC	required
args	Tuple	additional arguments for remaining un	()
name	str	name of the batch object	None
shuffle	bool	shuffle the order of jobs	False

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/quera.py

@beartype
def run(
    self,
    shots: int,
    RemoteTask: type[RemoteTaskType],
    args: Tuple[LiteralType, ...] = (),
    name: Optional[str] = None,
    use_experimental: bool = False,
    shuffle: bool = False,
    **kwargs,
) -> RemoteBatch:
    """Run the custom task and return the result.

    Args:
        shots (int): number of shots
        RemoteTask (type): type of the remote task, must subclass of CustomRemoteTaskABC
        args (Tuple): additional arguments for remaining un
        name (str): name of the batch object
        shuffle (bool): shuffle the order of jobs
    """
    if not callable(getattr(RemoteTask, "pull", None)):
        raise TypeError(
            f"{RemoteTask} must have a `pull` method for executing `run`."
        )

    batch = self.run_async(
        shots, RemoteTask, args, name, use_experimental, shuffle, **kwargs
    )
    batch.pull()
    return batch

run_async

run_async(
    shots: int,
    RemoteTask: type[RemoteTaskType],
    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.

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 .venv/lib/python3.12/site-packages/bloqade/analog/ir/routine/quera.py

@beartype
def run_async(
    self,
    shots: int,
    RemoteTask: type[RemoteTaskType],
    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_custom_batch(
        shots, RemoteTask, use_experimental, args, name
    )
    batch._submit(shuffle, **kwargs)
    return batch

submit

submit(
    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.

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 .venv/lib/python3.12/site-packages/bloqade/analog/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_single(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: 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.

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 .venv/lib/python3.12/site-packages/bloqade/analog/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