Local mode job class (#2057)

* Initial commit

* Fix lint

* Add properties

* Update qiskit_ibm_runtime/fake_provider/local_runtime_job.py

Co-authored-by: Takashi Imamichi <[email protected]>

* Add methods and properties

* Add reno

* Add test

---------

Co-authored-by: Takashi Imamichi <[email protected]>

qiskit_ibm_runtime/fake_provider/local_runtime_job.py

@@ -0,0 +1,105 @@
+# 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"],
+        inputs: dict,
+        *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._inputs = inputs
+        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 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()
+
+    def error_message(self) -> str:
+        """Returns the reason if the job failed."""
+        return ""
+
+    @property
+    def inputs(self) -> Dict:
+        """Return job input parameters."""
+        return self._inputs
+
+    @property
+    def session_id(self) -> str:
+        """Return the Session ID which would just be the job ID in local mode."""
+
+        return self._job_id
+
+    @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

qiskit_ibm_runtime/fake_provider/local_service.py

@@ -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
 
@@ -267,4 +268,14 @@ 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,
+            inputs=inputs,
+        )
+
+        return local_runtime_job

release-notes/unreleased/2057.feat.rst

@@ -0,0 +1,4 @@
+Jobs run in the local testing mode will now return an instance of a new class, 
+:class:`.LocalRuntimeJob`. This new class inherits from Qiskit's ``PrimitiveJob`` class 
+while adding the methods and properties found in :class:`.BaseRuntimeJob`. This way, running jobs 
+in the local testing mode will be more similar to running jobs on a real backend. 

test/unit/test_local_mode.py

@@ -25,6 +25,7 @@
 from qiskit.primitives.containers.data_bin import DataBin
 
 from qiskit_ibm_runtime.fake_provider import FakeManilaV2
+from qiskit_ibm_runtime.fake_provider.local_runtime_job import LocalRuntimeJob
 from qiskit_ibm_runtime import (
     Session,
     Batch,
@@ -160,3 +161,18 @@ def test_non_primitive(self, backend):
         session = Session(backend=backend)
         with self.assertRaisesRegex(ValueError, "Only sampler and estimator"):
             session._run(program_id="foo", inputs={})
+
+
+class TestLocalRuntimeJob(IBMTestCase):
+    """Class for testing local mode runtime jobs."""
+
+    def test_v2_sampler(self):
+        """Test V2 Sampler on a local backend."""
+        inst = SamplerV2(mode=FakeManilaV2())
+        job = inst.run(**get_primitive_inputs(inst))
+
+        self.assertIsInstance(job, LocalRuntimeJob)
+        self.assertTrue(job.metrics())
+        self.assertTrue(job.backend())
+        self.assertTrue(job.inputs)
+        self.assertEqual(job.usage(), 0)