GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/qdk

Public

mirrored from https://github.com/microsoft/qdkAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
v1.28.0

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

qdk/source/pip/qsharp/qre

source/pip/qsharp/qre/_application.py

172lines · modecode

RawDownload
Latest commit unavailable.
1# Copyright (c) Microsoft Corporation.
2# Licensed under the MIT License.
3
4from __future__ import annotations
5
6import types
7from abc import ABC, abstractmethod
8from concurrent.futures import ThreadPoolExecutor
9from types import NoneType
10from typing import (
11 ClassVar,
12 Generic,
13 Protocol,
14 TypeVar,
15 Generator,
16 get_type_hints,
17 cast,
18)
19
20from ._enumeration import _enumerate_instances
21from ._qre import Trace, EstimationResult
22from ._trace import TraceQuery
23
24
25class DataclassProtocol(Protocol):
26 __dataclass_fields__: ClassVar[dict]
27
28
29TraceParameters = TypeVar("TraceParameters", DataclassProtocol, types.NoneType)
30
31
32class Application(ABC, Generic[TraceParameters]):
33 """
34 An application defines a class of quantum computation problems along with a
35 method to generate traces for specific problem instances.
36
37 We distinguish between application and trace parameters. The application
38 parameters define which particular instance of the application we want to
39 consider. The trace parameters define how to generate a trace. They change
40 the specific way in which we solve the problem, but not the problem itself.
41
42 For example, in quantum cryptanalysis, the application parameters could
43 define the key size for an RSA prime product, while the trace parameters
44 define which algorithm to use to break the cryptography, as well as
45 parameters therein.
46 """
47
48 _parallel_traces: bool = True
49
50 @abstractmethod
51 def get_trace(self, parameters: TraceParameters) -> Trace:
52 """Return the trace corresponding to this application and parameters.
53
54 Args:
55 parameters (TraceParameters): The trace parameters.
56
57 Returns:
58 Trace: The trace for this application instance and parameters.
59 """
60
61 @staticmethod
62 def q(**kwargs) -> TraceQuery:
63 """Create a trace query for this application.
64
65 Args:
66 **kwargs: Domain overrides forwarded to trace parameter enumeration.
67
68 Returns:
69 TraceQuery: A trace query for this application type.
70 """
71 return TraceQuery(NoneType, **kwargs)
72
73 def context(self) -> _Context:
74 """Create a new enumeration context for this application."""
75 return _Context(self)
76
77 def post_process(
78 self, parameters: TraceParameters, estimation: EstimationResult
79 ) -> EstimationResult:
80 """Post-process an estimation result for a given set of trace parameters."""
81 return estimation
82
83 def enumerate_traces(
84 self,
85 **kwargs,
86 ) -> Generator[Trace, None, None]:
87 """Yield all traces of an application given its dataclass parameters.
88
89 Args:
90 **kwargs: Domain overrides forwarded to ``_enumerate_instances``.
91
92 Yields:
93 Trace: A trace for each enumerated set of trace parameters.
94 """
95
96 param_type = get_type_hints(self.__class__.get_trace).get("parameters")
97 if param_type is types.NoneType:
98 yield self.get_trace(None) # type: ignore
99 return
100
101 if isinstance(param_type, TypeVar):
102 for c in param_type.__constraints__:
103 if c is not types.NoneType:
104 param_type = c
105 break
106
107 if self._parallel_traces:
108 instances = list(_enumerate_instances(cast(type, param_type), **kwargs))
109 with ThreadPoolExecutor() as executor:
110 for trace in executor.map(self.get_trace, instances):
111 yield trace
112 else:
113 for instances in _enumerate_instances(cast(type, param_type), **kwargs):
114 yield self.get_trace(instances)
115
116 def enumerate_traces_with_parameters(
117 self,
118 **kwargs,
119 ) -> Generator[tuple[TraceParameters, Trace], None, None]:
120 """Yield (parameters, trace) pairs for an application.
121
122 Like ``enumerate_traces``, but each yielded trace is accompanied by the
123 trace parameters that were used to generate it.
124
125 Args:
126 **kwargs: Domain overrides forwarded to ``_enumerate_instances``.
127
128 Yields:
129 tuple[TraceParameters, Trace]: A pair of trace parameters and
130 the corresponding trace.
131 """
132
133 param_type = get_type_hints(self.__class__.get_trace).get("parameters")
134 if param_type is types.NoneType:
135 yield None, self.get_trace(None) # type: ignore
136 return
137
138 if isinstance(param_type, TypeVar):
139 for c in param_type.__constraints__:
140 if c is not types.NoneType:
141 param_type = c
142 break
143
144 if self._parallel_traces:
145 instances = list(_enumerate_instances(cast(type, param_type), **kwargs))
146 with ThreadPoolExecutor() as executor:
147 for instance, trace in zip(
148 instances, executor.map(self.get_trace, instances)
149 ):
150 yield instance, trace
151 else:
152 for instance in _enumerate_instances(cast(type, param_type), **kwargs):
153 yield instance, self.get_trace(instance)
154
155 def disable_parallel_traces(self):
156 """Disable parallel trace generation for this application."""
157 self._parallel_traces = False
158
159
160class _Context:
161 """Enumeration context wrapping an application instance."""
162
163 application: Application
164
165 def __init__(self, application: Application, **kwargs):
166 """Initialize the context for the given application.
167
168 Args:
169 application (Application): The application instance.
170 **kwargs: Additional keyword arguments (reserved for future use).
171 """
172 self.application = application
173