We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
IonQ Job should throw TimeoutError when wait time exceeded
**Description of the issue**

Currently, `Job.results()` throws a `RuntimeError` when the job execution exceeds the `timeout_seconds` wait time. It would be more accurate to throw a `TimeoutError` instead.

see cirq-ionq/cirq_ionq/job.py:

https://github.com/quantumlib/Cirq/blob/7759c05fd71673ca58559307c220b5b779bf5bb6/cirq-ionq/cirq_ionq/job.py#L202

**How to reproduce the issue**

```
import cirq
import cirq.ionq as ionq

API_KEY = 'tomyheart' # Replace with your IonQ API key

service = ionq.Service(api_key=API_KEY, 
                       default_target='simulator')

q0, q1 = cirq.LineQubit.range(2)
circuit = cirq.Circuit(
    cirq.X(q0)**0.5,             # Square root of X
    cirq.CX(q0, q1),              # CNOT
    cirq.measure(q0, q1, key='b') # Measure both qubits
)

# create a job
job = service.create_job(circuit, repetitions=100)
job.results(timeout_seconds=0.1) # trigger a timeout
```

<details>

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
/tmp/ipykernel_6204/2129298043.py in <module>
----> 1 job.results(timeout_seconds=0.01)

~/anaconda3/envs/my-env/lib/python3.7/site-packages/cirq_ionq/job.py in results(self, timeout_seconds, polling_seconds)
    199                 raise RuntimeError(f'Job failed. Error message: {error}')
    200             raise RuntimeError(
--> 201                 f'Job was not completed successful. Instead had status: {self.status()}'
    202             )
    203         # IonQ returns results in little endian, Cirq prefers to use big endian, so we convert.

RuntimeError: Job was not completed successful. Instead had status: ready

</details>

**Cirq version**

0.12.0

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `cirq-ionq/cirq_ionq/job.py`
Content:
```
1 # Copyright 2020 The Cirq Developers
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     https://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Represents a job created via the IonQ API."""
15 
16 import time
17 from typing import Dict, Sequence, Union, TYPE_CHECKING
18 
19 from cirq_ionq import ionq_exceptions, results
20 from cirq._doc import document
21 
22 import cirq
23 
24 if TYPE_CHECKING:
25     import cirq_ionq
26 
27 
28 def _little_endian_to_big(value: int, bit_count: int) -> int:
29     return cirq.big_endian_bits_to_int(
30         cirq.big_endian_int_to_bits(value, bit_count=bit_count)[::-1]
31     )
32 
33 
34 class Job:
35     """A job created on the IonQ API.
36 
37     Note that this is mutable, when calls to get status or results are made
38     the job updates itself to the results returned from the API.
39 
40     If a job is canceled or deleted, only the job id and the status remain
41     valid.
42     """
43 
44     TERMINAL_STATES = ('completed', 'canceled', 'failed', 'deleted')
45     document(
46         TERMINAL_STATES,
47         'States of the IonQ API job from which the job cannot transition. '
48         'Note that deleted can only exist in a return call from a delete '
49         '(subsequent calls will return not found).',
50     )
51 
52     NON_TERMINAL_STATES = ('ready', 'submitted', 'running')
53     document(
54         NON_TERMINAL_STATES, 'States of the IonQ API job which can transition to other states.'
55     )
56 
57     ALL_STATES = TERMINAL_STATES + NON_TERMINAL_STATES
58     document(ALL_STATES, 'All states that an IonQ API job can exist in.')
59 
60     UNSUCCESSFUL_STATES = ('canceled', 'failed', 'deleted')
61     document(
62         UNSUCCESSFUL_STATES,
63         'States of the IonQ API job when it was not successful and so does not have any '
64         'data associated with it beyond an id and a status.',
65     )
66 
67     def __init__(self, client: 'cirq_ionq.ionq_client._IonQClient', job_dict: dict):
68         """Construct an IonQJob.
69 
70         Users should not call this themselves. If you only know the `job_id`, use `get_job`
71         on `cirq_ionq.Service`.
72 
73         Args:
74             client: The client used for calling the API.
75             job_dict: A dict representing the response from a call to get_job on the client.
76         """
77         self._client = client
78         self._job = job_dict
79 
80     def _refresh_job(self):
81         """If the last fetched job is not terminal, gets the job from the API."""
82         if self._job['status'] not in self.TERMINAL_STATES:
83             self._job = self._client.get_job(self.job_id())
84 
85     def _check_if_unsuccessful(self):
86         if self.status() in self.UNSUCCESSFUL_STATES:
87             raise ionq_exceptions.IonQUnsuccessfulJobException(self.job_id(), self.status())
88 
89     def job_id(self) -> str:
90         """Returns the job id (UID) for the job.
91 
92         This is the id used for identifying the job by the API.
93         """
94         return self._job['id']
95 
96     def status(self) -> str:
97         """Gets the current status of the job.
98 
99         This will get a new job if the status of the job previously was determined to not be in
100         a terminal state. A full list of states is given in `cirq_ionq.IonQJob.ALL_STATES`.
101 
102         Raises:
103             IonQException: If the API is not able to get the status of the job.
104 
105         Returns:
106             The job status.
107         """
108         self._refresh_job()
109         return self._job['status']
110 
111     def target(self) -> str:
112         """Returns the target where the job is to be run, or was run.
113 
114         Returns:
115             'qpu' or 'simulator' depending on where the job was run or is running.
116 
117         Raises:
118             IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.
119             IonQException: If unable to get the status of the job from the API.
120         """
121         self._check_if_unsuccessful()
122         return self._job['target']
123 
124     def name(self) -> str:
125         """Returns the name of the job which was supplied during job creation.
126 
127         This is different than the `job_id`.
128 
129         Raises:
130             IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.
131             IonQException: If unable to get the status of the job from the API.
132         """
133         self._check_if_unsuccessful()
134         return self._job['name']
135 
136     def num_qubits(self) -> int:
137         """Returns the number of qubits for the job.
138 
139         Raises:
140             IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.
141             IonQException: If unable to get the status of the job from the API.
142         """
143         self._check_if_unsuccessful()
144         return int(self._job['qubits'])
145 
146     def repetitions(self) -> int:
147         """Returns the number of repetitions for the job.
148 
149         Raises:
150             IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.
151             IonQException: If unable to get the status of the job from the API.
152         """
153         self._check_if_unsuccessful()
154         return int(self._job['metadata']['shots'])
155 
156     def measurement_dict(self) -> Dict[str, Sequence[int]]:
157         """Returns a dictionary of measurement keys to target qubit index."""
158         measurement_dict: Dict[str, Sequence[int]] = {}
159         if 'metadata' in self._job:
160             full_str = ''.join(
161                 value
162                 for key, value in self._job['metadata'].items()
163                 if key.startswith('measurement')
164             )
165             if full_str == '':
166                 return measurement_dict
167             for key_value in full_str.split(chr(30)):
168                 key, value = key_value.split(chr(31))
169                 measurement_dict[key] = [int(t) for t in value.split(',')]
170         return measurement_dict
171 
172     # TODO(#3388) Add documentation for Raises.
173     # pylint: disable=missing-raises-doc
174     def results(
175         self, timeout_seconds: int = 7200, polling_seconds: int = 1
176     ) -> Union[results.QPUResult, results.SimulatorResult]:
177         """Polls the IonQ api for results.
178 
179         Args:
180             timeout_seconds: The total number of seconds to poll for.
181             polling_seconds: The interval with which to poll.
182 
183         Returns:
184             Either a `cirq_ionq.QPUResults` or `cirq_ionq.SimulatorResults` depending on whether
185             the job was running on an actual quantum processor or a simulator.
186 
187         Raises:
188             IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.
189             IonQException: If unable to get the results from the API.
190         """
191         time_waited_seconds = 0
192         while time_waited_seconds < timeout_seconds:
193             # Status does a refresh.
194             if self.status() in self.TERMINAL_STATES:
195                 break
196             time.sleep(polling_seconds)
197             time_waited_seconds += polling_seconds
198         if self.status() != 'completed':
199             if 'failure' in self._job and 'error' in self._job['failure']:
200                 error = self._job['failure']['error']
201                 raise RuntimeError(f'Job failed. Error message: {error}')
202             raise RuntimeError(
203                 f'Job was not completed successful. Instead had status: {self.status()}'
204             )
205         # IonQ returns results in little endian, Cirq prefers to use big endian, so we convert.
206         if self.target() == 'qpu':
207             repetitions = self.repetitions()
208             counts = {
209                 _little_endian_to_big(int(k), self.num_qubits()): int(repetitions * float(v))
210                 for k, v in self._job['data']['histogram'].items()
211             }
212             return results.QPUResult(
213                 counts=counts,
214                 num_qubits=self.num_qubits(),
215                 measurement_dict=self.measurement_dict(),
216             )
217         else:
218             probabilities = {
219                 _little_endian_to_big(int(k), self.num_qubits()): float(v)
220                 for k, v in self._job['data']['histogram'].items()
221             }
222             return results.SimulatorResult(
223                 probabilities=probabilities,
224                 num_qubits=self.num_qubits(),
225                 measurement_dict=self.measurement_dict(),
226                 repetitions=self.repetitions(),
227             )
228 
229     # pylint: enable=missing-raises-doc
230     def cancel(self):
231         """Cancel the given job.
232 
233         This mutates the job to only have a job id and status `canceled`.
234         """
235         self._job = self._client.cancel_job(job_id=self.job_id())
236 
237     def delete(self):
238         """Delete the given job.
239 
240         This mutates the job to only have a job id and status `deleted`. Subsequence attempts to
241         get the job with this job id will return not found.
242         """
243         self._job = self._client.delete_job(job_id=self.job_id())
244 
245     def __str__(self) -> str:
246         return f'cirq_ionq.Job(job_id={self.job_id()})'
247
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/cirq-ionq/cirq_ionq/job.py b/cirq-ionq/cirq_ionq/job.py
--- a/cirq-ionq/cirq_ionq/job.py
+++ b/cirq-ionq/cirq_ionq/job.py
@@ -199,8 +199,10 @@
             if 'failure' in self._job and 'error' in self._job['failure']:
                 error = self._job['failure']['error']
                 raise RuntimeError(f'Job failed. Error message: {error}')
+            if time_waited_seconds >= timeout_seconds:
+                raise TimeoutError(f'Job timed out after waiting {time_waited_seconds} seconds.')
             raise RuntimeError(
-                f'Job was not completed successful. Instead had status: {self.status()}'
+                f'Job was not completed successfully. Instead had status: {self.status()}'
             )
         # IonQ returns results in little endian, Cirq prefers to use big endian, so we convert.
         if self.target() == 'qpu':

{"golden_diff": "diff --git a/cirq-ionq/cirq_ionq/job.py b/cirq-ionq/cirq_ionq/job.py\n--- a/cirq-ionq/cirq_ionq/job.py\n+++ b/cirq-ionq/cirq_ionq/job.py\n@@ -199,8 +199,10 @@\n             if 'failure' in self._job and 'error' in self._job['failure']:\n                 error = self._job['failure']['error']\n                 raise RuntimeError(f'Job failed. Error message: {error}')\n+            if time_waited_seconds >= timeout_seconds:\n+                raise TimeoutError(f'Job timed out after waiting {time_waited_seconds} seconds.')\n             raise RuntimeError(\n-                f'Job was not completed successful. Instead had status: {self.status()}'\n+                f'Job was not completed successfully. Instead had status: {self.status()}'\n             )\n         # IonQ returns results in little endian, Cirq prefers to use big endian, so we convert.\n         if self.target() == 'qpu':\n", "issue": "IonQ Job should throw TimeoutError when wait time exceeded\n**Description of the issue**\r\n\r\nCurrently, `Job.results()` throws a `RuntimeError` when the job execution exceeds the `timeout_seconds` wait time. It would be more accurate to throw a `TimeoutError` instead.\r\n\r\nsee cirq-ionq/cirq_ionq/job.py:\r\n\r\nhttps://github.com/quantumlib/Cirq/blob/7759c05fd71673ca58559307c220b5b779bf5bb6/cirq-ionq/cirq_ionq/job.py#L202\r\n\r\n**How to reproduce the issue**\r\n\r\n```\r\nimport cirq\r\nimport cirq.ionq as ionq\r\n\r\nAPI_KEY = 'tomyheart' # Replace with your IonQ API key\r\n\r\nservice = ionq.Service(api_key=API_KEY, \r\n                       default_target='simulator')\r\n\r\nq0, q1 = cirq.LineQubit.range(2)\r\ncircuit = cirq.Circuit(\r\n    cirq.X(q0)**0.5,             # Square root of X\r\n    cirq.CX(q0, q1),              # CNOT\r\n    cirq.measure(q0, q1, key='b') # Measure both qubits\r\n)\r\n\r\n# create a job\r\njob = service.create_job(circuit, repetitions=100)\r\njob.results(timeout_seconds=0.1) # trigger a timeout\r\n```\r\n\r\n<details>\r\n\r\n---------------------------------------------------------------------------\r\nRuntimeError                              Traceback (most recent call last)\r\n/tmp/ipykernel_6204/2129298043.py in <module>\r\n----> 1 job.results(timeout_seconds=0.01)\r\n\r\n~/anaconda3/envs/my-env/lib/python3.7/site-packages/cirq_ionq/job.py in results(self, timeout_seconds, polling_seconds)\r\n    199                 raise RuntimeError(f'Job failed. Error message: {error}')\r\n    200             raise RuntimeError(\r\n--> 201                 f'Job was not completed successful. Instead had status: {self.status()}'\r\n    202             )\r\n    203         # IonQ returns results in little endian, Cirq prefers to use big endian, so we convert.\r\n\r\nRuntimeError: Job was not completed successful. Instead had status: ready\r\n\r\n</details>\r\n\r\n**Cirq version**\r\n\r\n0.12.0\n", "before_files": [{"content": "# Copyright 2020 The Cirq Developers\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Represents a job created via the IonQ API.\"\"\"\n\nimport time\nfrom typing import Dict, Sequence, Union, TYPE_CHECKING\n\nfrom cirq_ionq import ionq_exceptions, results\nfrom cirq._doc import document\n\nimport cirq\n\nif TYPE_CHECKING:\n    import cirq_ionq\n\n\ndef _little_endian_to_big(value: int, bit_count: int) -> int:\n    return cirq.big_endian_bits_to_int(\n        cirq.big_endian_int_to_bits(value, bit_count=bit_count)[::-1]\n    )\n\n\nclass Job:\n    \"\"\"A job created on the IonQ API.\n\n    Note that this is mutable, when calls to get status or results are made\n    the job updates itself to the results returned from the API.\n\n    If a job is canceled or deleted, only the job id and the status remain\n    valid.\n    \"\"\"\n\n    TERMINAL_STATES = ('completed', 'canceled', 'failed', 'deleted')\n    document(\n        TERMINAL_STATES,\n        'States of the IonQ API job from which the job cannot transition. '\n        'Note that deleted can only exist in a return call from a delete '\n        '(subsequent calls will return not found).',\n    )\n\n    NON_TERMINAL_STATES = ('ready', 'submitted', 'running')\n    document(\n        NON_TERMINAL_STATES, 'States of the IonQ API job which can transition to other states.'\n    )\n\n    ALL_STATES = TERMINAL_STATES + NON_TERMINAL_STATES\n    document(ALL_STATES, 'All states that an IonQ API job can exist in.')\n\n    UNSUCCESSFUL_STATES = ('canceled', 'failed', 'deleted')\n    document(\n        UNSUCCESSFUL_STATES,\n        'States of the IonQ API job when it was not successful and so does not have any '\n        'data associated with it beyond an id and a status.',\n    )\n\n    def __init__(self, client: 'cirq_ionq.ionq_client._IonQClient', job_dict: dict):\n        \"\"\"Construct an IonQJob.\n\n        Users should not call this themselves. If you only know the `job_id`, use `get_job`\n        on `cirq_ionq.Service`.\n\n        Args:\n            client: The client used for calling the API.\n            job_dict: A dict representing the response from a call to get_job on the client.\n        \"\"\"\n        self._client = client\n        self._job = job_dict\n\n    def _refresh_job(self):\n        \"\"\"If the last fetched job is not terminal, gets the job from the API.\"\"\"\n        if self._job['status'] not in self.TERMINAL_STATES:\n            self._job = self._client.get_job(self.job_id())\n\n    def _check_if_unsuccessful(self):\n        if self.status() in self.UNSUCCESSFUL_STATES:\n            raise ionq_exceptions.IonQUnsuccessfulJobException(self.job_id(), self.status())\n\n    def job_id(self) -> str:\n        \"\"\"Returns the job id (UID) for the job.\n\n        This is the id used for identifying the job by the API.\n        \"\"\"\n        return self._job['id']\n\n    def status(self) -> str:\n        \"\"\"Gets the current status of the job.\n\n        This will get a new job if the status of the job previously was determined to not be in\n        a terminal state. A full list of states is given in `cirq_ionq.IonQJob.ALL_STATES`.\n\n        Raises:\n            IonQException: If the API is not able to get the status of the job.\n\n        Returns:\n            The job status.\n        \"\"\"\n        self._refresh_job()\n        return self._job['status']\n\n    def target(self) -> str:\n        \"\"\"Returns the target where the job is to be run, or was run.\n\n        Returns:\n            'qpu' or 'simulator' depending on where the job was run or is running.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return self._job['target']\n\n    def name(self) -> str:\n        \"\"\"Returns the name of the job which was supplied during job creation.\n\n        This is different than the `job_id`.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return self._job['name']\n\n    def num_qubits(self) -> int:\n        \"\"\"Returns the number of qubits for the job.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return int(self._job['qubits'])\n\n    def repetitions(self) -> int:\n        \"\"\"Returns the number of repetitions for the job.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return int(self._job['metadata']['shots'])\n\n    def measurement_dict(self) -> Dict[str, Sequence[int]]:\n        \"\"\"Returns a dictionary of measurement keys to target qubit index.\"\"\"\n        measurement_dict: Dict[str, Sequence[int]] = {}\n        if 'metadata' in self._job:\n            full_str = ''.join(\n                value\n                for key, value in self._job['metadata'].items()\n                if key.startswith('measurement')\n            )\n            if full_str == '':\n                return measurement_dict\n            for key_value in full_str.split(chr(30)):\n                key, value = key_value.split(chr(31))\n                measurement_dict[key] = [int(t) for t in value.split(',')]\n        return measurement_dict\n\n    # TODO(#3388) Add documentation for Raises.\n    # pylint: disable=missing-raises-doc\n    def results(\n        self, timeout_seconds: int = 7200, polling_seconds: int = 1\n    ) -> Union[results.QPUResult, results.SimulatorResult]:\n        \"\"\"Polls the IonQ api for results.\n\n        Args:\n            timeout_seconds: The total number of seconds to poll for.\n            polling_seconds: The interval with which to poll.\n\n        Returns:\n            Either a `cirq_ionq.QPUResults` or `cirq_ionq.SimulatorResults` depending on whether\n            the job was running on an actual quantum processor or a simulator.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the results from the API.\n        \"\"\"\n        time_waited_seconds = 0\n        while time_waited_seconds < timeout_seconds:\n            # Status does a refresh.\n            if self.status() in self.TERMINAL_STATES:\n                break\n            time.sleep(polling_seconds)\n            time_waited_seconds += polling_seconds\n        if self.status() != 'completed':\n            if 'failure' in self._job and 'error' in self._job['failure']:\n                error = self._job['failure']['error']\n                raise RuntimeError(f'Job failed. Error message: {error}')\n            raise RuntimeError(\n                f'Job was not completed successful. Instead had status: {self.status()}'\n            )\n        # IonQ returns results in little endian, Cirq prefers to use big endian, so we convert.\n        if self.target() == 'qpu':\n            repetitions = self.repetitions()\n            counts = {\n                _little_endian_to_big(int(k), self.num_qubits()): int(repetitions * float(v))\n                for k, v in self._job['data']['histogram'].items()\n            }\n            return results.QPUResult(\n                counts=counts,\n                num_qubits=self.num_qubits(),\n                measurement_dict=self.measurement_dict(),\n            )\n        else:\n            probabilities = {\n                _little_endian_to_big(int(k), self.num_qubits()): float(v)\n                for k, v in self._job['data']['histogram'].items()\n            }\n            return results.SimulatorResult(\n                probabilities=probabilities,\n                num_qubits=self.num_qubits(),\n                measurement_dict=self.measurement_dict(),\n                repetitions=self.repetitions(),\n            )\n\n    # pylint: enable=missing-raises-doc\n    def cancel(self):\n        \"\"\"Cancel the given job.\n\n        This mutates the job to only have a job id and status `canceled`.\n        \"\"\"\n        self._job = self._client.cancel_job(job_id=self.job_id())\n\n    def delete(self):\n        \"\"\"Delete the given job.\n\n        This mutates the job to only have a job id and status `deleted`. Subsequence attempts to\n        get the job with this job id will return not found.\n        \"\"\"\n        self._job = self._client.delete_job(job_id=self.job_id())\n\n    def __str__(self) -> str:\n        return f'cirq_ionq.Job(job_id={self.job_id()})'\n", "path": "cirq-ionq/cirq_ionq/job.py"}], "after_files": [{"content": "# Copyright 2020 The Cirq Developers\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Represents a job created via the IonQ API.\"\"\"\n\nimport time\nfrom typing import Dict, Sequence, Union, TYPE_CHECKING\n\nfrom cirq_ionq import ionq_exceptions, results\nfrom cirq._doc import document\n\nimport cirq\n\nif TYPE_CHECKING:\n    import cirq_ionq\n\n\ndef _little_endian_to_big(value: int, bit_count: int) -> int:\n    return cirq.big_endian_bits_to_int(\n        cirq.big_endian_int_to_bits(value, bit_count=bit_count)[::-1]\n    )\n\n\nclass Job:\n    \"\"\"A job created on the IonQ API.\n\n    Note that this is mutable, when calls to get status or results are made\n    the job updates itself to the results returned from the API.\n\n    If a job is canceled or deleted, only the job id and the status remain\n    valid.\n    \"\"\"\n\n    TERMINAL_STATES = ('completed', 'canceled', 'failed', 'deleted')\n    document(\n        TERMINAL_STATES,\n        'States of the IonQ API job from which the job cannot transition. '\n        'Note that deleted can only exist in a return call from a delete '\n        '(subsequent calls will return not found).',\n    )\n\n    NON_TERMINAL_STATES = ('ready', 'submitted', 'running')\n    document(\n        NON_TERMINAL_STATES, 'States of the IonQ API job which can transition to other states.'\n    )\n\n    ALL_STATES = TERMINAL_STATES + NON_TERMINAL_STATES\n    document(ALL_STATES, 'All states that an IonQ API job can exist in.')\n\n    UNSUCCESSFUL_STATES = ('canceled', 'failed', 'deleted')\n    document(\n        UNSUCCESSFUL_STATES,\n        'States of the IonQ API job when it was not successful and so does not have any '\n        'data associated with it beyond an id and a status.',\n    )\n\n    def __init__(self, client: 'cirq_ionq.ionq_client._IonQClient', job_dict: dict):\n        \"\"\"Construct an IonQJob.\n\n        Users should not call this themselves. If you only know the `job_id`, use `get_job`\n        on `cirq_ionq.Service`.\n\n        Args:\n            client: The client used for calling the API.\n            job_dict: A dict representing the response from a call to get_job on the client.\n        \"\"\"\n        self._client = client\n        self._job = job_dict\n\n    def _refresh_job(self):\n        \"\"\"If the last fetched job is not terminal, gets the job from the API.\"\"\"\n        if self._job['status'] not in self.TERMINAL_STATES:\n            self._job = self._client.get_job(self.job_id())\n\n    def _check_if_unsuccessful(self):\n        if self.status() in self.UNSUCCESSFUL_STATES:\n            raise ionq_exceptions.IonQUnsuccessfulJobException(self.job_id(), self.status())\n\n    def job_id(self) -> str:\n        \"\"\"Returns the job id (UID) for the job.\n\n        This is the id used for identifying the job by the API.\n        \"\"\"\n        return self._job['id']\n\n    def status(self) -> str:\n        \"\"\"Gets the current status of the job.\n\n        This will get a new job if the status of the job previously was determined to not be in\n        a terminal state. A full list of states is given in `cirq_ionq.IonQJob.ALL_STATES`.\n\n        Raises:\n            IonQException: If the API is not able to get the status of the job.\n\n        Returns:\n            The job status.\n        \"\"\"\n        self._refresh_job()\n        return self._job['status']\n\n    def target(self) -> str:\n        \"\"\"Returns the target where the job is to be run, or was run.\n\n        Returns:\n            'qpu' or 'simulator' depending on where the job was run or is running.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return self._job['target']\n\n    def name(self) -> str:\n        \"\"\"Returns the name of the job which was supplied during job creation.\n\n        This is different than the `job_id`.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return self._job['name']\n\n    def num_qubits(self) -> int:\n        \"\"\"Returns the number of qubits for the job.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return int(self._job['qubits'])\n\n    def repetitions(self) -> int:\n        \"\"\"Returns the number of repetitions for the job.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the status of the job from the API.\n        \"\"\"\n        self._check_if_unsuccessful()\n        return int(self._job['metadata']['shots'])\n\n    def measurement_dict(self) -> Dict[str, Sequence[int]]:\n        \"\"\"Returns a dictionary of measurement keys to target qubit index.\"\"\"\n        measurement_dict: Dict[str, Sequence[int]] = {}\n        if 'metadata' in self._job:\n            full_str = ''.join(\n                value\n                for key, value in self._job['metadata'].items()\n                if key.startswith('measurement')\n            )\n            if full_str == '':\n                return measurement_dict\n            for key_value in full_str.split(chr(30)):\n                key, value = key_value.split(chr(31))\n                measurement_dict[key] = [int(t) for t in value.split(',')]\n        return measurement_dict\n\n    # TODO(#3388) Add documentation for Raises.\n    # pylint: disable=missing-raises-doc\n    def results(\n        self, timeout_seconds: int = 7200, polling_seconds: int = 1\n    ) -> Union[results.QPUResult, results.SimulatorResult]:\n        \"\"\"Polls the IonQ api for results.\n\n        Args:\n            timeout_seconds: The total number of seconds to poll for.\n            polling_seconds: The interval with which to poll.\n\n        Returns:\n            Either a `cirq_ionq.QPUResults` or `cirq_ionq.SimulatorResults` depending on whether\n            the job was running on an actual quantum processor or a simulator.\n\n        Raises:\n            IonQUnsuccessfulJob: If the job has failed, been canceled, or deleted.\n            IonQException: If unable to get the results from the API.\n        \"\"\"\n        time_waited_seconds = 0\n        while time_waited_seconds < timeout_seconds:\n            # Status does a refresh.\n            if self.status() in self.TERMINAL_STATES:\n                break\n            time.sleep(polling_seconds)\n            time_waited_seconds += polling_seconds\n        if self.status() != 'completed':\n            if 'failure' in self._job and 'error' in self._job['failure']:\n                error = self._job['failure']['error']\n                raise RuntimeError(f'Job failed. Error message: {error}')\n            if time_waited_seconds >= timeout_seconds:\n                raise TimeoutError(f'Job timed out after waiting {time_waited_seconds} seconds.')\n            raise RuntimeError(\n                f'Job was not completed successfully. Instead had status: {self.status()}'\n            )\n        # IonQ returns results in little endian, Cirq prefers to use big endian, so we convert.\n        if self.target() == 'qpu':\n            repetitions = self.repetitions()\n            counts = {\n                _little_endian_to_big(int(k), self.num_qubits()): int(repetitions * float(v))\n                for k, v in self._job['data']['histogram'].items()\n            }\n            return results.QPUResult(\n                counts=counts,\n                num_qubits=self.num_qubits(),\n                measurement_dict=self.measurement_dict(),\n            )\n        else:\n            probabilities = {\n                _little_endian_to_big(int(k), self.num_qubits()): float(v)\n                for k, v in self._job['data']['histogram'].items()\n            }\n            return results.SimulatorResult(\n                probabilities=probabilities,\n                num_qubits=self.num_qubits(),\n                measurement_dict=self.measurement_dict(),\n                repetitions=self.repetitions(),\n            )\n\n    # pylint: enable=missing-raises-doc\n    def cancel(self):\n        \"\"\"Cancel the given job.\n\n        This mutates the job to only have a job id and status `canceled`.\n        \"\"\"\n        self._job = self._client.cancel_job(job_id=self.job_id())\n\n    def delete(self):\n        \"\"\"Delete the given job.\n\n        This mutates the job to only have a job id and status `deleted`. Subsequence attempts to\n        get the job with this job id will return not found.\n        \"\"\"\n        self._job = self._client.delete_job(job_id=self.job_id())\n\n    def __str__(self) -> str:\n        return f'cirq_ionq.Job(job_id={self.job_id()})'\n", "path": "cirq-ionq/cirq_ionq/job.py"}]}