Source code for xcc.device
"""
This module contains the :class:`~xcc.Device` class.
"""
from __future__ import annotations
import calendar
from datetime import time
from typing import Any, Mapping, Optional, Sequence, Tuple
from .connection import Connection
from .util import cached_property
[docs]class Device:
"""Represents a device on the Xanadu Cloud.
Args:
target (str): target name of the device
connection (Connection): connection to the Xanadu Cloud
.. note::
For performance reasons, the properties of a device are lazily fetched
and stored in a cache. This cache can be cleared at any time by calling
:meth:`Device.clear`.
.. warning::
The :class:`xcc.Device` class transparently contacts the Xanadu Cloud
when an uncached device property is accessed. This means that requesting
a device property for the first time may take longer than expected.
**Example:**
The following example shows how to use the :class:`Device` class to query
various properties of the ``X8_01`` device on the Xanadu Cloud. First, a
connection is established to the Xanadu Cloud:
>>> import xcc
>>> connection = xcc.Connection.load()
Next, a reference to the ``X8_01`` device is created using the connection.
>>> device = xcc.Device(target="X8_01", connection=connection)
Finally, the certificate, specification, status, etc. of the ``X8_01``
device are retrieved by accessing the corresponding property of the device:
>>> device.certificate
{'device_url': ..., 'laser_wavelength_meters': 1.55270048e-06}
>>> device.specification
{'gate_parameters': ..., 'target': 'X8_01'}
>>> device.status
'online'
"""
[docs] @staticmethod
def list(connection: Connection, status: Optional[str] = None) -> Sequence[Device]:
"""Returns devices on the Xanadu Cloud.
Args:
connection (Connection): connection to the Xanadu Cloud
status (str, optional): optionally filter devices by status
Returns:
Sequence[Device]: devices on the Xanadu Cloud which match the status filter
"""
response = connection.request("GET", "/devices")
def include(details: Mapping[str, Any]) -> bool:
"""Returns ``True`` if a device with the given details should be
included in the response. Otherwise, ``False`` is returned.
"""
return status is None or details["status"] == status
devices = []
for details in filter(include, response.json()["data"]):
device = Device(target=details["target"], connection=connection)
device._details = details # pylint: disable=protected-access
devices.append(device)
return devices
def __init__(self, target: str, connection: Connection) -> None:
self._target = target
self._connection = connection
@property
def target(self) -> str:
"""Returns the target of a device."""
return self._target
@property
def overview(self) -> Mapping[str, Any]:
"""Returns an overview of a device.
Returns:
Mapping[str, Any]: mapping from field names to values for this
device as determined by the needs of a Xanadu Cloud user.
"""
return {"target": self.target, "status": self.status}
@cached_property
def certificate(self) -> Mapping[str, Any]:
"""Returns the certificate of a device.
A device certificate contains the current operating conditions of a
device and is periodically updated while the device is online.
Returns:
Mapping[str, Any]: certificate of this device
.. warning::
The structure of a certificate may vary from device to device.
"""
return self._connection.request("GET", f"/devices/{self.target}/certificate").json()
@cached_property
def specification(self) -> Mapping[str, Any]:
"""Returns the specification of a device.
Returns:
Mapping[str, Any]: specification of this device
"""
return self._connection.request("GET", f"/devices/{self.target}/specifications").json()
@property
def expected_uptime(self) -> Mapping[str, Optional[Tuple[time, time]]]:
"""Returns the expected uptime of a device.
Returns:
Mapping[str, Optional[Tuple[time, time]]]: mapping from weekdays to
optional time pairs where each pair represents when the device is
expected to come online and offline that day
"""
# Ensure that the iteration order is chronological.
expected_uptime = dict.fromkeys(map(str.lower, calendar.day_name))
for weekday, times in self._details["expected_uptime"].items():
expected_uptime[weekday] = tuple(map(time.fromisoformat, times))
return expected_uptime
@property
def status(self) -> str:
"""Returns the status of a device.
Returns:
str: status of this device ("online" or "offline")
"""
return self._details["status"]
@property
def up(self) -> bool: # pylint: disable=invalid-name
"""Returns whether a device is accepting jobs.
Returns:
bool: ``True`` iff the status of this device is "online"
"""
return self.status == "online"
@cached_property
def _details(self) -> Mapping[str, Any]:
"""Returns the details of a device.
Returns:
Mapping[str, Any]: mapping from field names to values for this
device as determined by the Xanadu Cloud device endpoint.
.. note::
These fields are not intended to be directly accessed by external
callers. Instead, they should be individually retrieved through
their associated public properties.
"""
return self._connection.request("GET", f"/devices/{self.target}").json()
def __repr__(self) -> str:
"""Returns a printable representation of a device."""
return f"<{self.__class__.__name__}: target={self.target}>"
[docs] def clear(self) -> None:
"""Clears the details, certificate, and specification caches of a device."""
del self._details
del self.certificate
del self.specification
_modules/xcc/device
Download Python script
Download Notebook
View on GitHub