1"""File with templates"""
2
3import json
4import logging
5import pickle
6from collections.abc import Callable
7from pathlib import Path
8from typing import Literal, get_args, overload
9
10from qiskit.primitives.containers import SamplerPubLike
11
12from qlauncher.base import Algorithm, Backend, Problem, Result
13from qlauncher.base.base import Model
14from qlauncher.problems.circuit import _Circuit
15from qlauncher.routines.circuits import CIRCUIT_FORMATS
16from qlauncher.routines.qiskit.algorithms.wrapper import CircuitRunner
17from qlauncher.routines.qiskit.utils import coerce_to_circuit_list
18
19
20def _extract_args(argtypes: list[tuple[str, type]], args, kwargs) -> dict[str, object]:
21 if len(args) > len(argtypes):
22 return {}
23 as_kwargs = []
24 for name, _ in argtypes[len(args) :]:
25 if name not in kwargs:
26 return {}
27 as_kwargs.append(kwargs[name])
28
29 result = {}
30
31 for expected, received in zip(argtypes, list(args) + as_kwargs):
32 name, wanted_type = expected
33 if not isinstance(received, wanted_type):
34 return {}
35 result[name] = received
36
37 return result
38
39
[docs]
40class QLauncher:
41 """
42 QLauncher class.
43
44 Qlauncher is used to run quantum algorithms on specific problem instances and backends.
45 It provides methods for binding parameters, preparing the problem, running the algorithm, and processing the results.
46
47 Attributes:
48 problem (Problem): The problem instance to be solved.
49 algorithm (Algorithm): The quantum algorithm to be executed.
50 backend (Backend, optional): The backend to be used for execution. Defaults to None.
51 path (str): The path to save the results. Defaults to 'results/'.
52 binding_params (dict or None): The parameters to be bound to the problem and algorithm. Defaults to None.
53 encoding_type (type): The encoding type to be used changing the class of the problem. Defaults to None.
54
55 Example of usage::
56
57 from qlauncher import QLauncher
58 from qlauncher.problems import MaxCut
59 from qlauncher.routines.qiskit import QAOA, QiskitBackend
60
61 problem = MaxCut(instance_name='default')
62 algorithm = QAOA()
63 backend = QiskitBackend('local_simulator')
64
65 launcher = QLauncher(problem, algorithm, backend)
66 result = launcher.process(save_pickle=True)
67 print(result)
68
69 """
70
71 @overload
72 def __init__(
73 self, problem: Problem | Model, algorithm: Algorithm, backend: Backend, /, *, logger: logging.Logger | None = None
74 ) -> None:
75 """
76 Create a QLauncher instance that solves a `problem` using a given `algorithm` on a `backend`.
77
78 Args:
79 problem (Problem): Problem to solve.
80 algorithm (Algorithm): Algorithm to use.
81 backend (Backend | None, optional): Backend to run on.
82 logger (logging.Logger | None, optional): Logger. Defaults to None.
83 """
84
85 @overload
86 def __init__(
87 self, circuit: SamplerPubLike | CIRCUIT_FORMATS, backend: Backend, /, *, shots: int = 1024, logger: logging.Logger | None = None
88 ) -> None:
89 """
90 Create a QLauncher instance that samples `circuit` on the `backend` for `shots` shots.
91
92 Args:
93 circuit (SamplerPubLike): Circuit or (circuit, params) to sample.
94 backend (Backend): Backend to run the circuit on.
95 shots (int, optional): Samples to draw. Defaults to 1024.
96 logger (logging.Logger | None, optional): Logger. Defaults to None.
97 """
98
99 @overload
100 def __init__(self, problem: Problem | Model, algorithm: Algorithm, /, *, logger: logging.Logger | None = None) -> None:
101 """
102 Create a QLauncher instance that solves a `problem` using a given workflow `algorithm`. Backend is None.
103
104 Args:
105 problem (Problem | Model): Problem to solve.
106 algorithm (Algorithm): Algorithm to use.
107 logger (logging.Logger | None, optional): Logger. Defaults to None.
108 """
109
110 def __init__(self, *args, **kwargs) -> None:
111 if len(args) == 3:
112 problem: Problem | Model = args[0]
113 algorithm: Algorithm = args[1]
114 backend: Backend = args[2]
115 elif len(args) == 2 and isinstance(args[0], Problem | Model):
116 problem: Problem | Model = args[0]
117 algorithm: Algorithm = args[1]
118 backend: Backend = Backend('')
119 elif len(args) == 2 and isinstance(args[0], (*get_args(CIRCUIT_FORMATS), *get_args(SamplerPubLike))):
120 problem, algorithm, backend = self._build_from_circuit(args[0], args[1], kwargs.get('shots', 1024))
121 else:
122 raise TypeError
123 self.problem: Problem | Model = problem
124 self.algorithm = algorithm
125 self.backend = backend
126
127 logger = kwargs.get('logger')
128 if logger is None:
129 logger = logging.getLogger('QLauncher')
130 self.logger = logger
131
132 self.result: Result | None = None
133
134 def _get_compatible_problem(self, **formatter_kwargs) -> Model:
135 input_format = self.algorithm.get_input_format()
136 if input_format is None:
137 raise TypeError
138 problem = self.problem
139 methods = self._bfs_search(problem, input_format)
140 if methods is None:
141 raise TypeError
142
143 if len(methods) == 0:
144 return problem
145
146 if isinstance(problem, Problem):
147 # The first method is the Problem -> Model formatter.
148 problem = methods[0](problem, **formatter_kwargs)
149
150 for method in methods[1:]:
151 problem = method(problem)
152
153 return problem
154
[docs]
155 def run(self, **formatter_kwargs) -> Result:
156 """
157 Finds proper formatter, and runs the algorithm on the problem with given backends.
158
159 Returns:
160 dict: The results of the algorithm execution.
161 """
162 self.result = self.algorithm.run(self._get_compatible_problem(**formatter_kwargs), self.backend)
163 self.logger.info('Algorithm ended successfully!')
164 return self.result
165
166 def _build_from_circuit(
167 self, circuit: SamplerPubLike | CIRCUIT_FORMATS, backend: Backend, shots: int
168 ) -> tuple[Model, Algorithm, Backend]:
169 return (_Circuit(coerce_to_circuit_list(circuit)[0]), CircuitRunner(shots), backend)
170
171 def _bfs_search(self, problem: Problem | Model, input_format: type[Model]) -> list[Callable[[Problem | Model], Model]] | None:
172 to_check: list[tuple[list, type[Problem] | type[Model]]] = [([], type(problem))]
173 visited: set = {type(problem)}
174 if isinstance(problem, input_format):
175 return []
176 while len(to_check) > 0:
177 parents, current = to_check.pop(0)
178 if current is input_format:
179 return parents
180 for child, method in current._mapping.items():
181 if isinstance(child, str):
182 child = Model._all_problems[child]
183 if child in visited:
184 continue
185 to_check.append((parents + [method], child))
186 return None
187
[docs]
188 def save(self, path: str | Path, save_format: Literal['pickle', 'txt', 'json'] = 'pickle') -> None:
189 """
190 Save last run result to file
191
192 Args:
193 path (str): File path.
194 save_format (Literal['pickle', 'txt', 'json'], optional): Save format. Defaults to 'pickle'.
195
196 Raises:
197 ValueError: When no result is available or an incorrect save format was chosen
198 """
199 if self.result is None:
200 raise ValueError('No result to save')
201
202 # if not os.path.isfile(path):
203 # path = os.path.join(path, f'result-{datetime.now().isoformat(sep="_").replace(":","_")}.{save_format}')
204
205 self.logger.info('Saving results to file: %s', str(path))
206 if save_format == 'pickle':
207 with open(path, mode='wb') as f:
208 pickle.dump(self.result, f)
209 elif save_format == 'json':
210 with open(path, mode='w', encoding='utf-8') as f:
211 json.dump(self.result.__dict__, f, default=fix_json)
212 elif save_format == 'txt':
213 with open(path, mode='w', encoding='utf-8') as f:
214 f.write(str(self.result))
215 else:
216 raise ValueError(f'format: {save_format} in not supported try: pickle, txt, csv or json')
217
218
[docs]
219def fix_json(o: object):
220 # if o.__class__.__name__ == 'SamplingVQEResult':
221 # parsed = self.algorithm.parse_samplingVQEResult(o, self._full_path)
222 # return parsed
223 if o.__class__.__name__ == 'complex128':
224 return repr(o)
225 print(f'Name of object {o.__class__} not known, returning None as a json encodable')
226 return None
