Qiskit Serverless のコンピューティングリソースとデータリソースを管理する

# If you include the preceding `%%writefile` command (visible only when you read this locally in a
# notebook), running this cell saves to disk rather than executing the code.

from qiskit_serverless import update_status, Job

# # If your function has a mapping stage, particularly application functions, you can set the status
# to "RUNNING: MAPPING" as follows:
update_status(Job.MAPPING)

# # While handling transpilation, error suppression, and so forth, you can set the status to
# "RUNNING: OPTIMIZING_FOR_HARDWARE":
update_status(Job.OPTIMIZING_HARDWARE)

# # After you submit jobs to IBM Quantum Compute Service, the underlying quantum job will be queued. You can set
# status to "RUNNING: WAITING_FOR_QPU":
update_status(Job.WAITING_QPU)

# # When the Quantum Compute job starts running on the QPU, set the following status
# "RUNNING: EXECUTING_QPU":
update_status(Job.EXECUTING_QPU)

## Once QPU is completed and post-processing has begun, set the status "RUNNING: POST_PROCESSING":
update_status(Job.POST_PROCESSING)

# If you include the preceding `%%writefile` command (visible only when you read this locally in a
# notebook), running this cell saves to disk rather than executing the code.

from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime import QiskitRuntimeService
from qiskit_serverless import distribute_task

service = QiskitRuntimeService()

@distribute_task(target={"cpu": 1})
def transpile_remote(circuit, optimization_level, backend):
    """
    Transpiles an abstract circuit (or list of circuits)
    into an ISA circuit for a given backend.
    """
    pass_manager = generate_preset_pass_manager(
        optimization_level=optimization_level,
        backend=service.backend(backend)
    )
    isa_circuit = pass_manager.run(circuit)
    return isa_circuit

# If you include the preceding `%%writefile` command
# (visible only when you read this locally in a
# notebook), running this cell saves to disk rather than
# executing the code.

from time import time
from qiskit_serverless import get, get_arguments, save_result, update_status, Job

# Get arguments
arguments = get_arguments()
circuit = arguments.get("circuit")
optimization_level = arguments.get("optimization_level")
backend = arguments.get("backend")

# If you include the preceding `%%writefile` command
# (visible only when you read this locally in a
# notebook), running this cell saves to disk rather than executing the code.

# Start distributed transpilation
update_status(Job.OPTIMIZING_HARDWARE)

start_time = time()
transpile_worker_references = [
    transpile_remote(circuit, optimization_level, backend)
    for circuit in arguments.get("circuit_list")
]

transpiled_circuits = get(transpile_worker_references)
end_time = time()

# If you include the preceding `%%writefile` command
# (visible only when you read this locally in a
# notebook), running this cell saves to disk rather than executing the code.

# Save result, with metadata
result = {
    "circuits": transpiled_circuits,
    "metadata": {
        "resource_usage": {
            "RUNNING: OPTIMIZING_FOR_HARDWARE": {
                "CPU_TIME": end_time - start_time,
                "QPU_TIME": 0,
            },
        }
    },
}

save_result(result)

# If you include the preceding `%%writefile` command
# (visible only when you read this locally in a
# notebook), running this cell saves to disk rather than executing the code.

@distribute_task(target={
    "cpu": 16,
    "mem": 2 * 1024 * 1024 * 1024
})
def transpile_remote(circuit, optimization_level, backend):
    return None

import tarfile
from qiskit_serverless import IBMServerlessClient

# Create a tar
filename = "transpile_demo.tar"
file = tarfile.open(filename, "w")
file.add("./source_files/transpile_remote.py")
file.close()

# Get a reference to a QiskitFunction
serverless = IBMServerlessClient()
transpile_remote_demo = next(
    program
    for program in serverless.list()
    if program.title == "transpile_remote_serverless"
)

# Upload the tar to Serverless data directory
serverless.file_upload(file=filename, function=transpile_remote_demo)

serverless.files(function=transpile_remote_demo)

# If you include the preceding `%%writefile` command
# (visible only when you read this locally in a
# notebook), running this cell saves to disk rather than executing the code.

import tarfile
from qiskit_serverless import IBMServerlessClient

# For `token`, use the 44-character API_KEY you created
# and saved from the IBM Quantum Platform Home dashboard
serverless = IBMServerlessClient(token="<YOUR_API_KEY>")
files = serverless.files()
demo_file = files[0]
downloaded_tar = serverless.file_download(demo_file)


with tarfile.open(downloaded_tar, 'r') as tar:
    tar.extractall()