# This code is part of Qiskit.
#
# (C) Copyright IBM 2021.
#
# 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.
"""Queue information for a job."""
import warnings
from datetime import datetime
from typing import Any, Optional, Union, Dict
import dateutil.parser
from ..utils import utc_to_local, duration_difference
from ..utils.utils import api_status_to_job_status
[docs]class QueueInfo:
"""Queue information for a job."""
_data = {} # type: Dict
def __init__(
self,
position_in_queue: Optional[int] = None,
status: Optional[str] = None,
estimated_start_time: Optional[Union[str, datetime]] = None,
estimated_completion_time: Optional[Union[str, datetime]] = None,
hub_priority: Optional[float] = None,
group_priority: Optional[float] = None,
project_priority: Optional[float] = None,
job_id: Optional[str] = None,
**kwargs: Any
) -> None:
"""QueueInfo constructor.
Args:
position: Position in the queue.
status: Job status.
estimated_start_time: Estimated start time for the job, in UTC.
estimated_complete_time: Estimated complete time for the job, in UTC.
hub_priority: Dynamic priority for the hub.
group_priority: Dynamic priority for the group.
project_priority: Dynamic priority for the project.
job_id: Job ID.
kwargs: Additional attributes.
"""
self.position = int(position_in_queue) if position_in_queue else None
self._status = status
if isinstance(estimated_start_time, str):
estimated_start_time = dateutil.parser.isoparse(estimated_start_time)
if isinstance(estimated_completion_time, str):
estimated_completion_time = dateutil.parser.isoparse(
estimated_completion_time
)
self._estimated_start_time_utc = estimated_start_time
self._estimated_complete_time_utc = estimated_completion_time
self.hub_priority = hub_priority
self.group_priority = group_priority
self.project_priority = project_priority
self.job_id = job_id
self._data = kwargs
def __repr__(self) -> str:
"""Return the string representation of ``QueueInfo``.
Note:
The estimated start and end time are displayed in local time
for convenience.
Returns:
A string representation of ``QueueInfo``.
Raises:
TypeError: If the `estimated_start_time` or `estimated_end_time`
value is not valid.
"""
status = (
api_status_to_job_status(self._status).name
if self._status
else self._get_value(self._status)
)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
est_start_time = (
self.estimated_start_time.isoformat()
if self.estimated_start_time
else None
)
est_complete_time = (
self.estimated_complete_time.isoformat()
if self.estimated_complete_time
else None
)
queue_info = [
"job_id='{}'".format(self.job_id),
"_status='{}'".format(status),
"estimated_start_time='{}'".format(est_start_time),
"estimated_complete_time='{}'".format(est_complete_time),
"position={}".format(self.position),
"hub_priority={}".format(self.hub_priority),
"group_priority={}".format(self.group_priority),
"project_priority={}".format(self.project_priority),
]
return "<{}({})>".format(self.__class__.__name__, ", ".join(queue_info))
def __getattr__(self, name: str) -> Any:
try:
return self._data[name]
except KeyError:
raise AttributeError("Attribute {} is not defined.".format(name)) from None
def _get_value(
self, value: Optional[Any], default_value: str = "unknown"
) -> Optional[Any]:
"""Return the input value if it exists or the default.
Returns:
The input value if it is not ``None``, else the input default value.
"""
return value or default_value
@property
def estimated_start_time(self) -> Optional[datetime]:
"""Return estimated start time in local time."""
if self._estimated_start_time_utc is None:
return None
return utc_to_local(self._estimated_start_time_utc)
@property
def estimated_complete_time(self) -> Optional[datetime]:
"""Return estimated complete time in local time."""
if self._estimated_complete_time_utc is None:
return None
return utc_to_local(self._estimated_complete_time_utc)