Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[WIP] Local mode job class #2057

Draft
wants to merge 5 commits into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 88 additions & 0 deletions qiskit_ibm_runtime/fake_provider/local_runtime_job.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
# This code is part of Qiskit.
#
# (C) Copyright IBM 2024.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

"""Qiskit runtime local mode job class."""

from typing import Any, Dict, Literal
from datetime import datetime

from qiskit.primitives.primitive_job import PrimitiveJob
from qiskit_ibm_runtime.models import BackendProperties
from .fake_backend import FakeBackendV2 # pylint: disable=cyclic-import


class LocalRuntimeJob(PrimitiveJob):
"""Job class for qiskit-ibm-runtime's local mode."""

def __init__( # type: ignore[no-untyped-def]
self,
future,
backend: FakeBackendV2,
primitive: Literal["sampler", "estimator"],
*args,
**kwargs,
) -> None:
"""LocalRuntimeJob constructor.

Args:
future: Thread executor the job is run on.
backend: The backend to run the primitive on.
"""
super().__init__(*args, **kwargs)
self._future = future
self._backend = backend
self._primitive = primitive
self._created = datetime.now()
self._running = datetime.now()
self._finished = datetime.now()

def metrics(self) -> Dict[str, Any]:
"""Return job metrics.

Returns:
A dictionary with job metrics including but not limited to the following:

* ``timestamps``: Timestamps of when the job was created, started running, and finished.
* ``usage``: Details regarding job usage, the measurement of the amount of
time the QPU is locked for your workload.
"""
return {
"bss": {"seconds": 0},
"usage": {"quantum_seconds": 0, "seconds": 0},
"timestamps": {
"created": self._created,
"running": self._running,
"finished": self._finished,
},
}

def backend(self) -> FakeBackendV2:
"""Return the backend where this job was executed."""
return self._backend

def useage(self) -> float:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
def useage(self) -> float:
def usage(self) -> float:

"""Return job usage in seconds."""
return 0

def properties(self) -> BackendProperties:
"""Return the backend properties for this job"""
return self._backend.properties()

@property
def creation_date(self) -> datetime:
"""Job creation date in local time."""
return self._created

@property
def primitive_id(self) -> str:
"""Primitive name."""
return self._primitive
12 changes: 11 additions & 1 deletion qiskit_ibm_runtime/fake_provider/local_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@

from .fake_backend import FakeBackendV2 # pylint: disable=cyclic-import
from .fake_provider import FakeProviderForBackendV2 # pylint: disable=unused-import, cyclic-import
from .local_runtime_job import LocalRuntimeJob
from ..ibm_backend import IBMBackend
from ..runtime_options import RuntimeOptions

Expand Down Expand Up @@ -267,4 +268,13 @@ def _run_backend_primitive_v2(
if options_copy:
warnings.warn(f"Options {options_copy} have no effect in local testing mode.")

return primitive_inst.run(**inputs)
primitive_job = primitive_inst.run(**inputs)

local_runtime_job = LocalRuntimeJob(
function=primitive_job._function,
future=primitive_job._future,
backend=backend,
primitive=primitive,
)

return local_runtime_job
Loading