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
22 def __init__( # pylint: disable=super-init-not-called
23 self,
24 n_loops: int = 1,
25 loop_lengths: list[int] | tuple[int, ...] | npt.NDArray[np.int_] | None = None,
26 postselected: bool | None = None,
27 postselection: bool = True,
28 postselection_threshold: int | None = None,
29 ip_address: str | None = None,
30 url: str | None = None,
31 machine: str | None = None,
32 secret_key: str = '',
33 **kwargs,
34 ):
35 TBIDevice.__init__( # pylint: disable=non-parent-init-called
36 self,
37 n_loops=n_loops,
38 loop_lengths=loop_lengths,
39 postselected=postselected,
40 postselection=postselection,
41 postselection_threshold=postselection_threshold,
42 ip_address=ip_address,
43 url=url,
44 machine=machine,
45 )
46
47 self.pt_kwargs = kwargs
48 self.secret_key = secret_key
49
50 self.headers = {
51 "Authorization": f"Bearer {self.token}",
52 "Content-Type": "application/json",
53 }
54 self.sample_async_flag = False
55
56 def _submit_job_async(
57 self,
58 input_state: list[int] | tuple[int, ...],
59 bs_angles: list[float] | tuple[float, ...],
60 n_samples: int,
61 ) -> str:
62 """Prepares and sends sample request to PT.
63
64 Args:
65 input_state: description of input modes. The left-most entry corresponds to the first mode entering the loop.
66 bs_angles: list of beam splitter angles
67 n_samples: number of samples to draw. Defaults to 1.
68 """
69 task = OrcaTask(input_state, bs_angles,
70 self.loop_lengths, self.secret_key, self.machine, n_samples,
71 self.postselection, self.postselection_threshold,
72 **self.pt_kwargs)
73 return task.uid
74
75 def _request_samples(
76 self,
77 input_state: list[int] | tuple[int, ...],
78 bs_angles: list[float] | tuple[float, ...],
79 n_samples: int,
80 save_dir: FILE_LIKE | None = None,
81 ) -> npt.NDArray[np.int_]:
82 """Prepares and sends sample request to PT.
83
84 Args:
85 input_state: description of input modes.
86 The left-most entry corresponds to the first mode entering the loop.
87 bs_angles: list of beam splitter angles
88 n_samples: number of samples to draw. Defaults to 1.
89 save_dir: Path to the directory in which to save results. If set to None the results are not saved. Defaults
90 to None.
91 """
92 task = OrcaTask(input_state, bs_angles,
93 self.loop_lengths, self.machine, self.secret_key, n_samples,
94 self.postselection, self.postselection_threshold,
95 **self.pt_kwargs)
96 result_json = task.results()
97 samples = result_json
98 samples = self._reformat_samples(samples)
99 return samples
