1""" PT tbi Adapter """
2from __future__ import annotations
3
4import os
5from typing import TYPE_CHECKING
6
7import numpy as np
8import numpy.typing as npt
9from typing_extensions import TypeAlias
10
11from ptseries.tbi.pt import PT
12from ptseries.tbi.tbi_abstract import TBIDevice
13from qailab.orca_api.networking import OrcaTask
14
15if TYPE_CHECKING:
16 FILE_LIKE: TypeAlias = str | os.PathLike
17
18
[docs]
19class PTAdapter(PT):
20 """ Adapter for PT tbi """
21 # pylint:disable=too-few-public-methods
22
23 def __init__( # pylint: disable=super-init-not-called
24 self,
25 n_loops: int = 1,
26 loop_lengths: list[int] | tuple[int, ...] | npt.NDArray[np.int_] | None = None,
27 postselected: bool | None = None,
28 postselection: bool = True,
29 postselection_threshold: int | None = None,
30 ip_address: str | None = None,
31 url: str | None = None,
32 machine: str | None = None,
33 secret_key: str = '',
34 **kwargs,
35 ):
36 TBIDevice.__init__( # pylint: disable=non-parent-init-called
37 self,
38 n_loops=n_loops,
39 loop_lengths=loop_lengths,
40 postselected=postselected,
41 postselection=postselection,
42 postselection_threshold=postselection_threshold,
43 ip_address=ip_address,
44 url=url,
45 machine=machine,
46 )
47
48 self.pt_kwargs = kwargs
49 self.secret_key = secret_key
50
51 self.headers = {
52 "Authorization": f"Bearer {self.token}",
53 "Content-Type": "application/json",
54 }
55 self.sample_async_flag = False
56
57 def _submit_job_async(
58 self,
59 input_state: list[int] | tuple[int, ...],
60 bs_angles: list[float] | tuple[float, ...],
61 n_samples: int,
62 ) -> str:
63 """Prepares and sends sample request to PT.
64
65 Args:
66 input_state: description of input modes. The left-most entry corresponds to the first mode entering the loop.
67 bs_angles: list of beam splitter angles
68 n_samples: number of samples to draw. Defaults to 1.
69 """
70 task = OrcaTask(input_state, bs_angles,
71 self.loop_lengths, self.secret_key, self.machine, n_samples,
72 self.postselection, self.postselection_threshold,
73 **self.pt_kwargs)
74 return task.uid
75
76 def _request_samples(
77 self,
78 input_state: list[int] | tuple[int, ...],
79 bs_angles: list[float] | tuple[float, ...],
80 n_samples: int,
81 # TODO: figure out if we should do anything with this arg
82 save_dir: FILE_LIKE | None = None, # pylint:disable=unused-argument
83 ) -> npt.NDArray[np.int_]:
84 """Prepares and sends sample request to PT.
85
86 Args:
87 input_state: description of input modes.
88 The left-most entry corresponds to the first mode entering the loop.
89 bs_angles: list of beam splitter angles
90 n_samples: number of samples to draw. Defaults to 1.
91 save_dir: Path to the directory in which to save results. If set to None the results are not saved. Defaults
92 to None.
93 """
94 task = OrcaTask(input_state, bs_angles,
95 self.loop_lengths, self.machine, self.secret_key, n_samples,
96 self.postselection, self.postselection_threshold,
97 **self.pt_kwargs)
98 result_json = task.results()
99 samples = result_json
100 samples = self._reformat_samples(samples)
101 return samples
