Source code for qailab.orca_api.networking

  1"""Methods for communicating with psnc quantum api"""
  2import time
  3import requests
  4
  5

[docs]
  6class OrcaTask:
  7    """
  8    Represents a single task for orca on psnc quantum api.
  9    """
 10
 11    API_BASE_URL = 'https://api.quantum.psnc.pl/api/client'
 12
 13    def __init__(
 14            self,
 15            input_state,
 16            bs_angles,
 17            loop_lengths,
 18            machine,
 19            auth_token,
 20            n_samples=200,
 21            postselection=False,
 22            postselection_threshold=None,
 23            **kwargs) -> None:
 24
 25        self.machine = machine
 26        self.auth = auth_token
 27        self.task_payload = {
 28            'input_state': input_state,
 29            'bs_angles': bs_angles,
 30            'n_samples': n_samples,
 31            'loop_lengths': loop_lengths,
 32            'postselection': postselection,
 33            'postselection_threshold': postselection_threshold,
 34            'machine': machine,
 35            'extra_options': kwargs
 36        }
 37
 38        self.uid: str
 39        self._job_ids = []
 40        self._results = []
 41
 42        self._created_time = None
 43
 44        self._submitted = False
 45
 46        self._create_on_remote()
 47
 48    def _full_url(self, rel):
 49        return f"{OrcaTask.API_BASE_URL}/{rel}"
 50
 51    def _get_headers(self):
 52        return {
 53            'Content-Type': 'application/json',
 54            'Authorization': f'Bearer {self.auth}'
 55        }
 56
 57    def _create_on_remote(self):
 58        response = requests.post(
 59            self._full_url('tasks'),
 60            headers=self._get_headers(),
 61            json={
 62                'machine': self.machine,
 63                'payload': self.task_payload
 64            },
 65            timeout=5
 66        )
 67
 68        response.raise_for_status()
 69
 70        response_data = response.json()
 71
 72        self.uid = response_data.get('uid', None)
 73        self._created_time = response_data.get('created', None)
 74
 75        if self.uid is None:
 76            raise ValueError('API did not return task UID')
 77
 78    def _try_get_results(self):
 79
 80        response = requests.get(
 81            self._full_url(f'tasks/{self.uid}/results'),
 82            headers=self._get_headers(),
 83            timeout=5
 84        )
 85
 86        if not response.ok:
 87            return {}
 88
 89        response_data = response.json()
 90
 91        return response_data
 92

[docs]
 93    def results(self) -> list[str]:
 94        """
 95        Get task results. Blocks the thread until results are available.
 96
 97        Returns:
 98            list[str]: List of bitstrings obtained from runs.
 99        """
100        if self._results != []:
101            return self._results
102
103        res = self._try_get_results()
104        while res == {}:
105            time.sleep(0.05)
106            res = self._try_get_results()
107
108        if not isinstance(res, list):
109            raise ValueError(f'invalid result type: {type(res)}, {res}')
110
111        self._results = res
112        return res

113
114    @property
115    def status(self):
116        """Task status"""
117        response = requests.get(
118            self._full_url(f'tasks/{self.uid}/status'),
119            headers=self._get_headers(),
120            timeout=5
121        )
122
123        response.raise_for_status()
124
125        response_data = response.json()
126
127        return response_data.get('status', 'Unknown')