[cirqflow] Quantum runtime skeleton - part 3 (#4584)

This contains a skeleton of an execution loop that consumes `QuantumExecutable` and saves `ExecutableResult`

cirq-google/cirq_google/__init__.py

@@ -133,6 +133,7 @@
     ExecutableResult,
     ExecutableGroupResult,
     QuantumRuntimeConfiguration,
+    execute,
 )
 
 from cirq_google import experimental

cirq-google/cirq_google/workflow/__init__.py

@@ -12,4 +12,5 @@
     ExecutableResult,
     ExecutableGroupResult,
     QuantumRuntimeConfiguration,
+    execute,
 )

cirq-google/cirq_google/workflow/quantum_executable.py

@@ -17,10 +17,10 @@
 import abc
 import dataclasses
 from dataclasses import dataclass
-from typing import Union, Tuple, Optional, Sequence, cast, Iterable, Dict, Any, List
+from typing import Union, Tuple, Optional, Sequence, cast, Dict, Any, List, Iterator
 
-from cirq import _compat, study
 import cirq
+from cirq import _compat, study
 
 
 class ExecutableSpec(metaclass=abc.ABCMeta):
@@ -232,7 +232,7 @@ def __init__(
     def __len__(self) -> int:
         return len(self.executables)
 
-    def __iter__(self) -> Iterable[QuantumExecutable]:
+    def __iter__(self) -> Iterator[QuantumExecutable]:
         yield from self.executables
 
     def __str__(self) -> str:

cirq-google/cirq_google/workflow/quantum_runtime.py

@@ -12,9 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Runtime information dataclasses that accompany execution of executables."""
+"""Runtime information dataclasses and execution of executables."""
 
 import dataclasses
+import os
+import uuid
 from typing import Any, Dict, Optional, List
 
 import cirq
@@ -23,6 +25,7 @@
 from cirq_google.workflow._abstract_engine_processor_shim import AbstractEngineProcessorShim
 from cirq_google.workflow.quantum_executable import (
     ExecutableSpec,
+    QuantumExecutableGroup,
 )
 
 
@@ -133,3 +136,79 @@ def _json_dict_(self) -> Dict[str, Any]:
 
     def __repr__(self) -> str:
         return _compat.dataclass_repr(self, namespace='cirq_google')
+
+
+def execute(
+    rt_config: QuantumRuntimeConfiguration,
+    executable_group: QuantumExecutableGroup,
+    base_data_dir: str = ".",
+) -> ExecutableGroupResult:
+    """Execute a `cg.QuantumExecutableGroup` according to a `cg.QuantumRuntimeConfiguration`.
+
+    Args:
+        rt_config: The `cg.QuantumRuntimeConfiguration` specifying how to execute
+            `executable_group`.
+        executable_group: The `cg.QuantumExecutableGroup` containing the executables to execute.
+        base_data_dir: A filesystem path to write data. We write
+            "{base_data_dir}/{run_id}/ExecutableGroupResult.json.gz"
+            containing the `cg.ExecutableGroupResult` as well as one file
+            "{base_data_dir}/{run_id}/ExecutableResult.{i}.json.gz" per `cg.ExecutableResult` as
+            each executable result becomes available.
+
+    Returns:
+        The `cg.ExecutableGroupResult` containing all data and metadata for an execution.
+
+    Raises:
+        NotImplementedError: If an executable uses the `params` field or anything other than
+            a BitstringsMeasurement measurement field.
+        ValueError: If `base_data_dir` is not a valid directory.
+    """
+    # run_id defaults logic.
+    if rt_config.run_id is None:
+        run_id = str(uuid.uuid4())
+    else:
+        run_id = rt_config.run_id
+
+    # base_data_dir handling.
+    if not base_data_dir:
+        # coverage: ignore
+        raise ValueError("Please provide a non-empty `base_data_dir`.")
+
+    os.makedirs(f'{base_data_dir}/{run_id}', exist_ok=False)
+
+    # Results object that we will fill in in the main loop.
+    exegroup_result = ExecutableGroupResult(
+        runtime_configuration=rt_config,
+        shared_runtime_info=SharedRuntimeInfo(run_id=run_id),
+        executable_results=list(),
+    )
+    cirq.to_json_gzip(exegroup_result, f'{base_data_dir}/{run_id}/ExecutableGroupResult.json.gz')
+
+    # Loop over executables.
+    sampler = rt_config.processor.get_sampler()
+    n_executables = len(executable_group)
+    print()
+    for i, exe in enumerate(executable_group):
+        runtime_info = RuntimeInfo(execution_index=i)
+
+        if exe.params != tuple():
+            raise NotImplementedError("Circuit params are not yet supported.")
+
+        circuit = exe.circuit
+
+        if not hasattr(exe.measurement, 'n_repetitions'):
+            raise NotImplementedError("Only `BitstringsMeasurement` are supported.")
+
+        sampler_run_result = sampler.run(circuit, repetitions=exe.measurement.n_repetitions)
+
+        exe_result = ExecutableResult(
+            spec=exe.spec,
+            runtime_info=runtime_info,
+            raw_data=sampler_run_result,
+        )
+        cirq.to_json_gzip(exe_result, f'{base_data_dir}/{run_id}/ExecutableResult.{i}.json.gz')
+        exegroup_result.executable_results.append(exe_result)
+        print(f'\r{i+1} / {n_executables}', end='', flush=True)
+    print()
+
+    return exegroup_result

cirq-google/cirq_google/workflow/quantum_runtime_test.py

@@ -11,13 +11,19 @@
 # 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.
+import glob
+import re
+import uuid
 from dataclasses import dataclass
+from typing import List
+
+import numpy as np
+import pytest
 
 import cirq
 import cirq_google as cg
-import numpy as np
 from cirq_google.workflow._abstract_engine_processor_shim import AbstractEngineProcessorShim
-from cirq_google.workflow.quantum_executable_test import _get_example_spec
+from cirq_google.workflow.quantum_executable_test import _get_quantum_executables, _get_example_spec
 
 
 @dataclass
@@ -120,3 +126,30 @@ def test_executable_group_result(tmpdir):
     cg_assert_equivalent_repr(egr)
     assert len(egr.executable_results) == 3
     _assert_json_roundtrip(egr, tmpdir)
+
+
+@pytest.mark.parametrize('run_id', ['unit_test_runid', None])
+def test_execute(tmpdir, run_id):
+    rt_config = cg.QuantumRuntimeConfiguration(processor=_MockEngineProcessor(), run_id=run_id)
+    executable_group = cg.QuantumExecutableGroup(_get_quantum_executables())
+    returned_exegroup_result = cg.execute(
+        rt_config=rt_config, executable_group=executable_group, base_data_dir=tmpdir
+    )
+    actual_run_id = returned_exegroup_result.shared_runtime_info.run_id
+    if run_id is not None:
+        assert run_id == actual_run_id
+    else:
+        assert isinstance(uuid.UUID(actual_run_id), uuid.UUID)
+    fns = glob.glob(f'{tmpdir}/{actual_run_id}/ExecutableGroupResult.json.gz')
+    assert len(fns) == 1
+    exegroup_result: cg.ExecutableGroupResult = _cg_read_json_gzip(fns[0])
+
+    fns = glob.glob(f'{tmpdir}/{actual_run_id}/ExecutableResult.*.json.gz')
+    fns = sorted(
+        fns, key=lambda s: int(re.search(r'ExecutableResult\.(\d+)\.json\.gz$', s).group(1))
+    )
+    assert len(fns) == 3
+    exe_results: List[cg.ExecutableResult] = [_cg_read_json_gzip(fn) for fn in fns]
+
+    exegroup_result.executable_results = exe_results
+    assert returned_exegroup_result == exegroup_result