Change Sampler methods to return Sequence instead of List (quantumlib…

…#4807)

Sequence is a covariant type (Sequence[A] is a subclass of Sequence[B]
when A is a subclass of B) while List is invariant (List[A] is not a
subclass of List[B] even in A is a subclass of B). Returning
Sequence[cirq.Result] thus allows returning subtypes of cirq.Result,
while List[cirq.Result] does not.

Review: @viathor

cirq-aqt/cirq_aqt/aqt_sampler.py

@@ -25,7 +25,7 @@
 import json
 import time
 import uuid
-from typing import List, Union, Tuple, Dict, cast
+from typing import cast, Dict, List, Sequence, Tuple, Union
 
 import numpy as np
 from requests import put
@@ -179,7 +179,7 @@ def _send_json(
 
     def run_sweep(
         self, program: cirq.AbstractCircuit, params: cirq.Sweepable, repetitions: int = 1
-    ) -> List[cirq.Result]:
+    ) -> Sequence[cirq.Result]:
         """Samples from the given Circuit.
 
         In contrast to run, this allows for sweeping over different parameter

cirq-core/cirq/experiments/single_qubit_readout_calibration_test.py

@@ -11,7 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from typing import List
+from typing import Sequence
 import pytest
 
 import numpy as np
@@ -48,7 +48,7 @@ def run_sweep(
         program: 'cirq.AbstractCircuit',
         params: cirq.Sweepable,
         repetitions: int = 1,
-    ) -> List[cirq.Result]:
+    ) -> Sequence[cirq.Result]:
         results = self.simulator.run_sweep(program, params, repetitions)
         for result in results:
             for bits in result.measurements.values():

cirq-core/cirq/sim/clifford/stabilizer_sampler.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Dict, List
+from typing import Dict, List, Sequence
 
 import numpy as np
 
@@ -40,7 +40,7 @@ def run_sweep(
         program: 'cirq.AbstractCircuit',
         params: 'cirq.Sweepable',
         repetitions: int = 1,
-    ) -> List['cirq.Result']:
+    ) -> Sequence['cirq.Result']:
         results: List[cirq.Result] = []
         for param_resolver in cirq.to_resolvers(params):
             resolved_circuit = cirq.resolve_parameters(program, param_resolver)

cirq-core/cirq/sim/mux.py

@@ -17,7 +17,7 @@
 Filename is a reference to multiplexing.
 """
 
-from typing import List, Optional, Type, Union, cast, TYPE_CHECKING
+from typing import cast, List, Optional, Sequence, Type, TYPE_CHECKING, Union
 
 import numpy as np
 
@@ -169,7 +169,7 @@ def sample_sweep(
     repetitions: int = 1,
     dtype: Type[np.number] = np.complex64,
     seed: 'cirq.RANDOM_STATE_OR_SEED_LIKE' = None,
-) -> List['cirq.Result']:
+) -> Sequence['cirq.Result']:
     """Runs the supplied Circuit, mimicking quantum hardware.
 
     In contrast to run, this allows for sweeping over different parameter

cirq-core/cirq/sim/simulator.py

@@ -31,19 +31,19 @@
 import collections
 from typing import (
     Any,
+    Callable,
+    cast,
     Dict,
+    Generic,
     Iterator,
     List,
+    Optional,
     Sequence,
+    Set,
     Tuple,
-    Union,
-    Optional,
     TYPE_CHECKING,
-    Set,
-    cast,
-    Callable,
     TypeVar,
-    Generic,
+    Union,
 )
 
 import numpy as np
@@ -73,7 +73,7 @@ def run_sweep(
         program: 'cirq.AbstractCircuit',
         params: 'cirq.Sweepable',
         repetitions: int = 1,
-    ) -> List['cirq.Result']:
+    ) -> Sequence['cirq.Result']:
         return list(self.run_sweep_iter(program, params, repetitions))
 
     def run_sweep_iter(

cirq-core/cirq/work/sampler.py

@@ -14,10 +14,10 @@
 """Abstract base class for things sampling quantum circuits."""
 
 import abc
-from typing import List, Optional, TYPE_CHECKING, Union, Dict, FrozenSet, Tuple
-from typing import Sequence
+from typing import Dict, FrozenSet, List, Optional, Sequence, Tuple, TYPE_CHECKING, Union
 
 import pandas as pd
+
 from cirq import study, ops
 from cirq.work.observable_measurement import (
     measure_observables,
@@ -148,7 +148,7 @@ def run_sweep(
         program: 'cirq.AbstractCircuit',
         params: 'cirq.Sweepable',
         repetitions: int = 1,
-    ) -> List['cirq.Result']:
+    ) -> Sequence['cirq.Result']:
         """Samples from the given Circuit.
 
         In contrast to run, this allows for sweeping over different parameter
@@ -187,7 +187,7 @@ async def run_sweep_async(
         program: 'cirq.AbstractCircuit',
         params: 'cirq.Sweepable',
         repetitions: int = 1,
-    ) -> List['cirq.Result']:
+    ) -> Sequence['cirq.Result']:
         """Asynchronously sweeps and samples from the given Circuit.
 
         By default, this method invokes `run_sweep` synchronously and simply
@@ -211,7 +211,7 @@ def run_batch(
         programs: Sequence['cirq.AbstractCircuit'],
         params_list: Optional[List['cirq.Sweepable']] = None,
         repetitions: Union[int, List[int]] = 1,
-    ) -> List[List['cirq.Result']]:
+    ) -> Sequence[Sequence['cirq.Result']]:
         """Runs the supplied circuits.
 
         Each circuit provided in `programs` will pair with the optional
@@ -277,7 +277,7 @@ def sample_expectation_values(
         num_samples: int,
         params: 'cirq.Sweepable' = None,
         permit_terminal_measurements: bool = False,
-    ) -> List[List[float]]:
+    ) -> Sequence[Sequence[float]]:
         """Calculates estimated expectation values from samples of a circuit.
 
         Please see also `cirq.work.measure_observables` for more control over how to measure

cirq-core/cirq/work/sampler_test.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Tests for cirq.Sampler."""
-from typing import List
+from typing import Sequence
 
 import pytest
 
@@ -225,7 +225,7 @@ def run_sweep(
             program: 'cirq.AbstractCircuit',
             params: 'cirq.Sweepable',
             repetitions: int = 1,
-        ) -> List['cirq.Result']:
+        ) -> Sequence['cirq.Result']:
             results = np.zeros((repetitions, 1), dtype=bool)
             for idx in range(repetitions // 4):
                 results[idx][0] = 1

cirq-google/cirq_google/api/v2/results.py

@@ -19,6 +19,7 @@
     Iterator,
     List,
     Optional,
+    Sequence,
     Set,
 )
 from collections import OrderedDict
@@ -162,7 +163,7 @@ def results_to_proto(
 def results_from_proto(
     msg: result_pb2.Result,
     measurements: List[MeasureInfo] = None,
-) -> List[List[cirq.Result]]:
+) -> Sequence[Sequence[cirq.Result]]:
     """Converts a v2 result proto into List of list of trial results.
 
     Args:
@@ -185,7 +186,7 @@ def results_from_proto(
 def _trial_sweep_from_proto(
     msg: result_pb2.SweepResult,
     measure_map: Dict[str, MeasureInfo] = None,
-) -> List[cirq.Result]:
+) -> Sequence[cirq.Result]:
     """Converts a SweepResult proto into List of list of trial results.
 
     Args:

cirq-google/cirq_google/engine/abstract_job.py

@@ -14,7 +14,7 @@
 """A helper for jobs that have been created on the Quantum Engine."""
 
 import abc
-from typing import Dict, Iterator, List, Optional, overload, Tuple, TYPE_CHECKING
+from typing import Dict, Iterator, List, Optional, overload, Sequence, Tuple, TYPE_CHECKING
 
 import cirq
 from cirq_google.engine.client import quantum
@@ -161,7 +161,7 @@ def delete(self) -> Optional[bool]:
         """Deletes the job and result, if any."""
 
     @abc.abstractmethod
-    def batched_results(self) -> List[List[cirq.Result]]:
+    def batched_results(self) -> Sequence[Sequence[cirq.Result]]:
         """Returns the job results, blocking until the job is complete.
 
         This method is intended for batched jobs.  Instead of flattening
@@ -170,11 +170,11 @@ def batched_results(self) -> List[List[cirq.Result]]:
         """
 
     @abc.abstractmethod
-    def results(self) -> List[cirq.Result]:
+    def results(self) -> Sequence[cirq.Result]:
         """Returns the job results, blocking until the job is complete."""
 
     @abc.abstractmethod
-    def calibration_results(self) -> List['calibration_result.CalibrationResult']:
+    def calibration_results(self) -> Sequence['calibration_result.CalibrationResult']:
         """Returns the results of a run_calibration() call.
 
         This function will fail if any other type of results were returned.
@@ -189,7 +189,7 @@ def __getitem__(self, item: int) -> cirq.Result:
         pass
 
     @overload
-    def __getitem__(self, item: slice) -> List[cirq.Result]:
+    def __getitem__(self, item: slice) -> Sequence[cirq.Result]:
         pass
 
     def __getitem__(self, item):

cirq-google/cirq_google/engine/abstract_local_job_test.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """A helper for jobs that have been created on the Quantum Engine."""
-from typing import List, Optional, Tuple
+from typing import Optional, Sequence, Tuple
 import datetime
 import cirq
 
@@ -40,13 +40,13 @@ def cancel(self) -> None:
     def delete(self) -> None:
         pass
 
-    def batched_results(self) -> List[List[cirq.Result]]:
+    def batched_results(self) -> Sequence[Sequence[cirq.Result]]:
         return []  # coverage: ignore
 
-    def results(self) -> List[cirq.Result]:
+    def results(self) -> Sequence[cirq.Result]:
         return []  # coverage: ignore
 
-    def calibration_results(self) -> List[CalibrationResult]:
+    def calibration_results(self) -> Sequence[CalibrationResult]:
         return []  # coverage: ignore
 
 

cirq-google/cirq_google/engine/engine_job.py

@@ -15,7 +15,7 @@
 import datetime
 import time
 
-from typing import Dict, Iterator, List, Optional, overload, Tuple, TYPE_CHECKING
+from typing import Dict, Iterator, List, Optional, overload, Sequence, Tuple, TYPE_CHECKING
 
 import cirq
 from cirq_google.engine import abstract_job, calibration, engine_client
@@ -79,9 +79,9 @@ def __init__(
         self.job_id = job_id
         self.context = context
         self._job = _job
-        self._results: Optional[List[cirq.Result]] = None
-        self._calibration_results: Optional[CalibrationResult] = None
-        self._batched_results: Optional[List[List[cirq.Result]]] = None
+        self._results: Optional[Sequence[cirq.Result]] = None
+        self._calibration_results: Optional[Sequence[CalibrationResult]] = None
+        self._batched_results: Optional[Sequence[Sequence[cirq.Result]]] = None
         self.result_type = result_type
 
     def id(self) -> str:
@@ -258,11 +258,11 @@ def delete(self) -> None:
         """Deletes the job and result, if any."""
         self.context.client.delete_job(self.project_id, self.program_id, self.job_id)
 
-    def batched_results(self) -> List[List[cirq.Result]]:
+    def batched_results(self) -> Sequence[Sequence[cirq.Result]]:
         """Returns the job results, blocking until the job is complete.
 
         This method is intended for batched jobs.  Instead of flattening
-        results into a single list, this will return a List[Result]
+        results into a single list, this will return a Sequence[Result]
         for each circuit in the batch.
         """
         self.results()
@@ -288,7 +288,7 @@ def _wait_for_result(self):
         )
         return response.result
 
-    def results(self) -> List[cirq.Result]:
+    def results(self) -> Sequence[cirq.Result]:
         """Returns the job results, blocking until the job is complete."""
         import cirq_google.engine.engine as engine_base
 
@@ -315,7 +315,7 @@ def results(self) -> List[cirq.Result]:
                 raise ValueError(f'invalid result proto version: {result_type}')
         return self._results
 
-    def calibration_results(self):
+    def calibration_results(self) -> Sequence[CalibrationResult]:
         """Returns the results of a run_calibration() call.
 
         This function will fail if any other type of results were returned
@@ -334,24 +334,25 @@ def calibration_results(self):
                 metrics = calibration.Calibration(layer.metrics)
                 message = layer.error_message or None
                 token = layer.token or None
+                ts: Optional[datetime.datetime] = None
                 if layer.valid_until_ms > 0:
                     ts = datetime.datetime.fromtimestamp(layer.valid_until_ms / 1000)
-                else:
-                    ts = None
                 cal_results.append(CalibrationResult(layer.code, message, token, ts, metrics))
             self._calibration_results = cal_results
         return self._calibration_results
 
     @classmethod
-    def _get_batch_results_v2(cls, results: v2.batch_pb2.BatchResult) -> List[List[cirq.Result]]:
+    def _get_batch_results_v2(
+        cls, results: v2.batch_pb2.BatchResult
+    ) -> Sequence[Sequence[cirq.Result]]:
         trial_results = []
         for result in results.results:
             # Add a new list for the result
             trial_results.append(_get_job_results_v2(result))
         return trial_results
 
     @classmethod
-    def _flatten(cls, result) -> List[cirq.Result]:
+    def _flatten(cls, result) -> Sequence[cirq.Result]:
         return [res for result_list in result for res in result_list]
 
     def __iter__(self) -> Iterator[cirq.Result]:
@@ -363,7 +364,7 @@ def __getitem__(self, item: int) -> cirq.Result:
         pass
 
     @overload
-    def __getitem__(self, item: slice) -> List[cirq.Result]:
+    def __getitem__(self, item: slice) -> Sequence[cirq.Result]:
         pass
 
     def __getitem__(self, item):
@@ -403,7 +404,7 @@ def _deserialize_run_context(
     raise ValueError(f'unsupported run_context type: {run_context_type}')
 
 
-def _get_job_results_v1(result: v1.program_pb2.Result) -> List[cirq.Result]:
+def _get_job_results_v1(result: v1.program_pb2.Result) -> Sequence[cirq.Result]:
     trial_results = []
     for sweep_result in result.sweep_results:
         sweep_repetitions = sweep_result.repetitions
@@ -421,7 +422,7 @@ def _get_job_results_v1(result: v1.program_pb2.Result) -> List[cirq.Result]:
     return trial_results
 
 
-def _get_job_results_v2(result: v2.result_pb2.Result) -> List[cirq.Result]:
+def _get_job_results_v2(result: v2.result_pb2.Result) -> Sequence[cirq.Result]:
     sweep_results = v2.results_from_proto(result)
     # Flatten to single list to match to sampler api.
     return [trial_result for sweep_result in sweep_results for trial_result in sweep_result]

cirq-google/cirq_google/engine/engine_sampler.py

@@ -53,7 +53,7 @@ def run_sweep(
         program: Union[cirq.AbstractCircuit, 'cirq_google.EngineProgram'],
         params: cirq.Sweepable,
         repetitions: int = 1,
-    ) -> List[cirq.Result]:
+    ) -> Sequence[cirq.Result]:
         if isinstance(program, engine.EngineProgram):
             job = program.run_sweep(
                 params=params, repetitions=repetitions, processor_ids=self._processor_ids
@@ -70,10 +70,10 @@ def run_sweep(
 
     def run_batch(
         self,
-        programs: Sequence['cirq.AbstractCircuit'],
+        programs: Sequence[cirq.AbstractCircuit],
         params_list: Optional[List[cirq.Sweepable]] = None,
         repetitions: Union[int, List[int]] = 1,
-    ) -> List[List[cirq.Result]]:
+    ) -> Sequence[Sequence[cirq.Result]]:
         """Runs the supplied circuits.
 
         In order to gain a speedup from using this method instead of other run